Мы разработали сервис для удобного просмотра заблокированных в СПБ бирже бумаг используя интеграцию с 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.