Обмен заблокированных активов СБП биржа
Мы разработали сервис для удобного просмотра заблокированных в СПБ бирже бумаг используя интеграцию с Tinkoff Invest API.
Собираем данных о замороженных позициях, сравниваем их с доступным к обмену списку от организатора торгов ООО “Инвестиционная Палата” и отображаем финальную сумму конвертации.
Сервис для проверки активов в брокере Тинькофф: https://844.artydev.ru/
Введение
Важно иметь в виду, что иностранные акции, которые были заблокированы после наложенных на СПБ Биржу в 2023 году санкций, не могут участвовать в продаже по указу № 844 — даже если они включены в список на сайте организатора торгов.
Вот как устанавливается цена продажи заблокированных активов:
- Организатор торгов установил минимальную цену продажи для каждого актива — это цена актива на иностранных биржах 22 марта 2024 года, но рассчитанная в рублях по курсу ЦБ РФ. Посмотреть минимальные цены активов на сайте организатора торгов.
- С 3 июня по 5 июля организатор торгов будет собирать от нерезидентов заявки на покупку заблокированных активов.
- Конечная цена продажи будет определена на основе спроса и предложения.
Если в момент продажи активов будет повышенный спрос от нерезидентов, активы могут быть проданы даже выше установленной минимальной цены. А если спроса от нерезидентов не окажется, активы просто не будут проданы — даже с дисконтом.
Как получить токен
- Перейдите в настройки профиля Тинькофф Инвестиции по ссылке: https://www.tinkoff.ru/invest/settings/
- Авторизуйтесь в системе, если это требуется.
- Выпустите токен TINKOFF INVEST API для биржи. Возможно, система попросит вас авторизоваться еще раз. Не беспокойтесь, это необходимо для подключения робота к торговой платформе.
- Скопируйте токен и сохраните его. Токен отображается только один раз, просмотреть его позже не получится. Тем не менее вы можете выпускать неограниченное количество токенов.
Разработка
Основная обработка токена состоит из 3 функций:
- Сбор данных о заблокированных позициях
- Трансформация в pandas DataFrame
- Расчет итоговых сумм
def get_user_portfolio(token: str) -> list:
info = []
with Client(token) as client:
accounts = client.users.get_accounts()
for account in accounts.accounts:
logger.info(f"Check account: {account.name}")
portfolio = client.operations.get_portfolio(account_id=account.id)
for pos in portfolio.positions:
if pos.blocked is True:
price = quotation_to_decimal(pos.current_price)
data = {
"figi": pos.figi,
"quantity": pos.quantity_lots.units,
"price": price,
"total": price * pos.quantity_lots.units,
"currency": pos.current_price.currency.upper(),
"acc": account.name,
}
logger.info(f"Blocked position: {data}")
info.append(data)
return info
def convert_portfolio_to_df(data: list) -> pd.DataFrame:
df = pd.DataFrame(data)
df["total"] = df["total"].astype(float)
df["quantity"] = df["quantity"].astype(int)
df["price"] = df["price"].astype(float)
df["zfigi"] = df["figi"].str[4:]
cb_data["zfigi"] = cb_data["figi"].str[4:]
merge_df = df.merge(cb_data, how="left", on="zfigi", suffixes=("", "_cb"))
columns = ["figi", "quantity", "name", "price", "total", "currency", "acc", "isin", "_type", "price_cb"]
merge_df = merge_df[columns]
merge_df["total_cb_price"] = merge_df["quantity"] * merge_df["price_cb"]
merge_df.loc[merge_df['isin'].isnull(), 'is_exchange'] = 'N'
merge_df.loc[merge_df['isin'].notnull(), 'is_exchange'] = 'Y'
merge_df.loc[(df['quantity'] > 0) & (merge_df['total'] == 0), 'is_used'] = 'Y'
merge_df.loc[(df['quantity'] > 0) & (merge_df['total'] > 0), 'is_used'] = 'N'
return merge_df
def convert_df_to_dict(data: pd.DataFrame) -> dict:
user_data = {
"rows": data.to_dict("records"),
"total_currency": data[data.is_exchange == "Y"].total.sum(),
"total_rub": data[data.is_exchange == "Y"].total_cb_price.sum(),
"total_rub_used": data[data.is_used == "Y"].total_cb_price.sum(),
"status": "success"
}
logger.info(f"Prepare to final response: {user_data}")
return user_data
Эндпоинт в API сервисе на получение данных
@app.post('/api/v1/exchange_info', status_code=201, response_model=UserResponse)
async def get_spb_info(data: UserRequest):
request_data = data.model_dump()
token = request_data["token"]
try:
data = get_user_portfolio(token)
df = convert_portfolio_to_df(data)
response_data = convert_df_to_dict(df)
return response_data
except UnauthenticatedError:
logger.error(f"Error with token: {token}. {traceback.format_exc()}")
return {
"rows": [],
"total_currency": 0.0,
"total_rub": 0.0,
"status": "failed",
}
Большое спасибо всем за внимание! Если вам интересны подобные рассуждения - подписывайтесь на мой канал artydev & Co.