Обмен заблокированных активов СБП биржа

2 minute read
projects
,

img

Мы разработали сервис для удобного просмотра заблокированных в СПБ бирже бумаг используя интеграцию с Tinkoff Invest API.
Собираем данных о замороженных позициях, сравниваем их с доступным к обмену списку от организатора торгов ООО “Инвестиционная Палата” и отображаем финальную сумму конвертации.

Сервис для проверки активов в брокере Тинькофф: https://844.artydev.ru/

Введение

Важно иметь в виду, что иностранные акции, которые были заблокированы после наложенных на СПБ Биржу в 2023 году санкций, не могут участвовать в продаже по указу № 844 — даже если они включены в список на сайте организатора торгов.

Вот как устанавливается цена продажи заблокированных активов:

  • Организатор торгов установил минимальную цену продажи для каждого актива — это цена актива на иностранных биржах 22 марта 2024 года, но рассчитанная в рублях по курсу ЦБ РФ. Посмотреть минимальные цены активов на сайте организатора торгов.
  • С 3 июня по 5 июля организатор торгов будет собирать от нерезидентов заявки на покупку заблокированных активов.
  • Конечная цена продажи будет определена на основе спроса и предложения.

Если в момент продажи активов будет повышенный спрос от нерезидентов, активы могут быть проданы даже выше установленной минимальной цены. А если спроса от нерезидентов не окажется, активы просто не будут проданы — даже с дисконтом.

Как получить токен

  1. Перейдите в настройки профиля Тинькофф Инвестиции по ссылке: https://www.tinkoff.ru/invest/settings/
  2. Авторизуйтесь в системе, если это требуется.
  3. Выпустите токен TINKOFF INVEST API для биржи. Возможно, система попросит вас авторизоваться еще раз. Не беспокойтесь, это необходимо для подключения робота к торговой платформе.
  4. Скопируйте токен и сохраните его. Токен отображается только один раз, просмотреть его позже не получится. Тем не менее вы можете выпускать неограниченное количество токенов.

img

Разработка

Основная обработка токена состоит из 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",
        }

Большое спасибо всем за внимание! Если вам интересны подобные рассуждения - подписывайтесь на мой канал  Telegram_logo artydev & Co.