Тинькофф инвестиции пульс api wrapper

6 minute read

img

Введение

В этом посте я хотел бы уделить внимание созданию полноценного python-пакета на основе wrapper’a (обертки) над API социальной сети Пульс в Тинькофф инвестициях
Пакет будет включаться в себя 3 базовых метода:

  • Получить базовую информацию по пользователю
  • Получить список постов по id пользователя
  • Получить список постов по тикеру

Результат

Итоговый python пакет: https://pypi.org/project/tpulse/
Репозиторий в github: https://github.com/meanother/tpulse-py

Разработка пакета

Нам понадобятся 2 python-модуля:

  • httpx (аналог requests)
  • fake_useragent (подмена user-agent в заголовках запроса)
pip install httpx fake_useragent

class ClientBase:

class ClientBase:
    """Base class for API client"""

    def __init__(self, base_url: str):
        headers = {
            "Content-type": "application/json",
            "Accept": "application/json",
            "User-agent": ua,
        }
        data = {"appName": "invest", "origin": "web", "platform": "web"}
        self._client = httpx.Client(base_url=base_url, headers=headers, params=data)

    def __enter__(self) -> "ClientBase":
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.close()

    def close(self):
        """Close network connections"""
        self._client.close()

    def _get(self, url, data, timeout=settings.TIMEOUT_SEC):
        """GET request to Dadata API"""
        response = self._client.get(url, params=data, timeout=timeout)
        response.raise_for_status()
        return response.json()

class UserClient:

class UserClient(ClientBase):
    """User class for tpulse api"""

    BASE_URL = "https://www.tinkoff.ru/api/invest-gw/social/v1/"

    def __init__(self):
        super().__init__(base_url=self.BASE_URL)

    def user_info(self, name: str) -> Optional[Dict]:
        """get user info by username"""
        url = "profile/nickname/%s" % name
        response = self._get(url, data=None)
        return response["payload"] if response["status"] == "Ok" else None

    def user_posts(self, user_id: str, cursor: int, **kwargs) -> Optional[Dict]:
        """get user posts by user id"""
        url = "profile/%s/post" % user_id
        data = {"limit": 30, "cursor": cursor}
        data.update(kwargs)
        response = self._get(url, data)
        return response["payload"] if response["status"] == "Ok" else None

class PostClient:

class PostClient(ClientBase):
    """Ticker class for tpulse api"""

    BASE_URL = "https://www.tinkoff.ru/api/invest-gw/social/v1/"

    def __init__(self):
        super().__init__(base_url=self.BASE_URL)

    def posts(self, ticker: str, cursor: int, **kwargs) -> Optional[Dict]:
        """get post info by ticker"""
        url = "post/instrument/%s" % ticker
        data = {"limit": 30, "cursor": cursor}
        data.update(kwargs)
        response = self._get(url, data)
        return response["payload"] if response["status"] == "Ok" else None

class PulseClient:

class PulseClient:
    """Sync client for tpulse api"""

    def __init__(self):
        self._user = UserClient()
        self._post = PostClient()

    def test(self):
        """test func"""
        pass

    def get_user_info(self, name: str) -> Optional[Dict]:
        """Get user info"""
        return self._user.user_info(name)

    def get_posts_by_user_id(
        self, user_id: str, cursor: int = 999999999, **kwargs
    ) -> Optional[Dict]:
        """Collect last 30 posts for user"""
        return self._user.user_posts(user_id, cursor, **kwargs)

    def get_posts_by_ticker(
        self, ticker: str, cursor: int = 999999999, **kwargs
    ) -> Optional[Dict]:
        """Collect last 30 posts for ticker"""
        return self._post.posts(ticker, cursor, **kwargs)

    def __enter__(self) -> "PulseClient":
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.close()

    def close(self):
        """Close network connections"""
        self._user.close()
        self._post.close()

Инициализация пакета

Для начала создадим шаблон пакета с помощью утилиты flit
Установим ее:

pip install flit

В интерактивном режиме заполним базовое описание пакета

flit init

По итогу получаем следующий файл: pyproject.toml

Flit создал файл с метаданными проекта pyproject.toml. В нем уже есть все необходимое для публикации пакета в публичном репозитории — PyPi.
Далее необходимо зарегистрироваться в тестовом и основном PyPi репозиториях:

Для дальнейшего использования потребуется создать файл в корне диска ~/.pypirc

[distutils]
index-servers =
    pypi
    testpypi

[testpypi]
repository = https://test.pypi.org/legacy/
username: artydev
password: your_password

[pypi]
username: artydev
password: your_password

Публикация пакета

# В тестомовм репозитории
flit publish --repository pypitest

# В основном репозитории
flit publish

Качество кода и автотесты

Установим необходимые пакеты и создадим конфигурационный файл tox.ini

pip install black coverage flake8 mccabe mypy pylint pytest tox
[gh-actions]
python =
    3.7: py37
    3.8: py38
    3.9: py39

[tox]
isolated_build = True
envlist = python3.7,py38,py39

[testenv]
deps =
    black
    coverage
    flake8
    mccabe
    pylint
    pytest
    httpx
    fake_useragent
    pytest_httpx
commands =
    black tpulse
    flake8 tpulse
    pylint tpulse
    coverage erase
    coverage run --include=tpulse/* -m pytest -ra
    coverage report -m
    coverage xml

tox.ini

Запуск проверок осуществляется с помощью команды tox -e py38 для конкретной версии, либо tox для полной проверки

py38 run-test: commands[5] | coverage report -m
Name                    Stmts   Miss  Cover   Missing
-----------------------------------------------------
tpulse/__init__.py          2      0   100%
tpulse/settings.py          1      0   100%
tpulse/sync_client.py      67      6    91%   25, 28, 32, 42-44
-----------------------------------------------------
TOTAL                      70      6    91%
py38 run-test: commands[6] | coverage xml
Wrote XML report to coverage.xml

py38: commands succeeded
congratulations :)

Сборка в облаке

GitHub Actions позволяет запускать сборки и автотесты в облаке, используя docker-контейнеры

  • Покрытие кода через Codecov
  • Качество кода через Codeclimate

Добавим конфиг для GitHub Actions в .github/workflows/build.yml:

name: build

on:
    push:
        branches: [main]
    pull_request:
        branches: [main]
    workflow_dispatch:

jobs:
    build:
        runs-on: ubuntu-latest
        strategy:
            matrix:
                python-version: [3.7, 3.8, 3.9]

        env:
            USING_COVERAGE: "3.9"

        steps:
            - name: Checkout sources
              uses: actions/checkout@v2

            - name: Set up Python
              uses: actions/setup-python@v2
              with:
                  python-version: ${{ matrix.python-version }}

            - name: Install dependencies
              run: |
                  python -m pip install --upgrade pip
                  python -m pip install black coverage flake8 flit mccabe mypy pylint pytest tox tox-gh-actions httpx fake_useragent pytest_httpx
            - name: Run tox
              run: |
                  python -m tox
            - name: Upload coverage to Codecov
              uses: codecov/codecov-action@v1
              if: contains(env.USING_COVERAGE, matrix.python-version)
              with:
                  fail_ci_if_error: true

build.yml

Для отображения бирок добавим следующие строки в README.md

[![PyPI Version][pypi-image]][pypi-url]
[![Build Status][build-image]][build-url]
[![Code Coverage][coverage-image]][coverage-url]
[![Code Quality][quality-image]][quality-url]

[pypi-image]: https://img.shields.io/pypi/v/tpulse
[pypi-url]: https://pypi.org/project/tpulse/
[build-image]: https://github.com/meanother/tpulse-py/actions/workflows/build.yml/badge.svg
[build-url]: https://github.com/meanother/tpulse-py/actions/workflows/build.yml
[coverage-image]: https://codecov.io/gh/meanother/tpulse-py/branch/main/graph/badge.svg
[coverage-url]: https://codecov.io/gh/nameanotherlgeon/tpulse-py
[quality-image]: https://api.codeclimate.com/v1/badges/ca8f259b0ad93f1f28ed/maintainability
[quality-url]: https://codeclimate.com/github/meanother/tpulse-py

Установка и использование

Наш модуль опубликован в pip репозитории, откуда мы можем его смело установить:

pip install tpulse

Базовая информация по пользователю

>>> from tpulse import TinkoffPulse
>>> from pprint import pp
>>> pulse = TinkoffPulse()
>>>
>>> user_info = pulse.get_user_info("tomcapital")
>>> pp(user_info)
{'id': 'bfbc4cc2-7f98-472e-8f5f-a14bdd6fc4db',
 'type': 'personal',
 'nickname': 'TomCapital',
 'status': 'open',
 'image': '22ac448f-e271-463c-beb1-f035c7987f17',
 'block': False,
 'description': 'Эксклюзивная аналитика тут: https://t.me/tomcapital\n'
                '\n'
                'Связь: https://t.me/TomCapCat\n'
                '\n'
                'growth stocks strategy\n'
                '\n'
                'Ты должен изучить правила игры. Затем начать играть лучше, '
                'чем кто-либо другой.',
 'followersCount': 39704,
 'followingCount': 13,
 'isLead': False,
 'serviceTags': [{'id': 'popular'}],
 'statistics': {'totalAmountRange': {'lower': 3000000, 'upper': None},
                'yearRelativeYield': -5.68,
                'monthOperationsCount': 98},
 'subscriptionDomains': None,
 'popularHashtags': [],
 'donationActive': True,
 'isVisible': True,
 'baseTariffCategory': 'unauthorized',
 'strategies': [{'id': 'a48ee1fc-4eaa-47a3-a75c-a362d3c95cdf',
                 'title': 'Tactical Investing',
                 'riskProfile': 'moderate',
                 'relativeYield': 3.93,
                 'baseCurrency': 'usd',
                 'score': 4,
                 'portfolioValues': [...],
                 'characteristics': [{'id': 'recommended-base-money-position-quantity',
                                      'value': '1\xa0100 $',
                                      'subtitle': 'советуем вложить'},
                                     {'id': 'slaves-count',
                                      'value': '111',
                                      'subtitle': 'подписаны'}]},
                {'id': 'ff41c693-78dd-4c2e-b566-858770d6d2e0',
                 'title': 'Aggressive investing',
                 'riskProfile': 'aggressive',
                 'relativeYield': -8.19,
                 'baseCurrency': 'usd',
                 'score': 3,
                 'portfolioValues': [...],
                 'characteristics': [{'id': 'recommended-base-money-position-quantity',
                                      'value': '1\xa0000 $',
                                      'subtitle': 'советуем вложить'},
                                     {'id': 'slaves-count',
                                      'value': '17',
                                      'subtitle': 'подписаны'}]}]}

Список постов по id пользователя

>>> user_posts = pulse.get_posts_by_user_id("bfbc4cc2-7f98-472e-8f5f-a14bdd6fc4db")
>>> pp(user_posts)
...
>>> pp(user_posts["items"][0])
{'id': '2ab5457c-aa9d-4a9b-b7ea-7af49459f0f9',
 'text': 'Множество акций испытали массивную коррекцию за последние несколько '
         'недель, особенно это касается growth-историй (компаний, чей '
         'потенциал и денежные потоки должны раскрыться в будущем). На '
         'фондовый рынок обрушилась целая лавина плохих новостей (высказывания '
         'Пауэлла, тейперинг, Omicron и тд), и, на мой взгляд, мы увидели '
         'некую чрезмерную реакцию рынка.\n'
         '\n'
         'Часто, когда фондовый рынок заранее корректируется и закладывает те '
         'или иные негативные события в оценку активов, то уже непосредственно '
         'по факту наступления этих самых событий, рынок, как правило, '
         'успевает переварить их, и, наоборот, раллирует. Особенно, если '
         'случилась избыточная или даже паническая реакция на негатив.\n'
         '\n'
         'Марко Коланович, главный стратег JPMorgan, оценивает вероятность '
         'шорт-сквиз ралли в ближайшие недели, как высокую, и я, пожалуй, буду '
         'придерживаться такой же точки зрения.',
 'likesCount': 42,
 'commentsCount': 10,
 'isLiked': False,
 'inserted': '2021-12-22T15:22:38.016128+03:00',
 'isEditable': False,
 'instruments': [],
 'profiles': [],
 'serviceTags': [],
 'profileId': 'bfbc4cc2-7f98-472e-8f5f-a14bdd6fc4db',
 'nickname': 'TomCapital',
 'image': '22ac448f-e271-463c-beb1-f035c7987f17',
 'postImages': [],
 'hashtags': [],
 'owner': {'id': 'bfbc4cc2-7f98-472e-8f5f-a14bdd6fc4db',
           'nickname': 'TomCapital',
           'image': '22ac448f-e271-463c-beb1-f035c7987f17',
           'donationActive': False,
           'block': False,
           'serviceTags': [{'id': 'popular'}]},
 'reactions': {'totalCount': 42,
               'myReaction': None,
               'counters': [{'type': 'like', 'count': 42}]},
 'content': {'type': 'simple',
             'text': '',
             'instruments': [],
             'hashtags': [],
             'profiles': [],
             'images': [],
             'strategies': []},
 'baseTariffCategory': 'unauthorized',
 'isBookmarked': False,
 'status': 'published'}

Список постов по тикеру

>>> ticker_posts = pulse.get_posts_by_ticker("AAPL")
>>> pp(ticker_posts)
...
>>> pp(ticker_posts["items"][5])
{'id': '320b8e15-fe8c-46e9-b29b-12ef278be135',
 'text': '{$AAPL} продажу поставил на 176 $',
 'likesCount': 0,
 'commentsCount': 6,
 'isLiked': False,
 'inserted': '2021-12-23T11:54:50.603445+03:00',
 'isEditable': False,
 'instruments': [{'type': 'share',
                  'ticker': 'AAPL',
                  'lastPrice': 176.02,
                  'currency': 'usd',
                  'image': 'US0378331005.png',
                  'briefName': 'Apple',
                  'dailyYield': None,
                  'relativeDailyYield': 0.0,
                  'price': 175.34,
                  'relativeYield': 0.39}],
 'profiles': [],
 'serviceTags': [],
 'profileId': '436a1012-3c5d-4c84-879b-a4e434f43230',
 'nickname': 'TNEO',
 'image': 'fc85fbc9-ef4a-4045-905d-bd6fb581689c',
 'postImages': [],
 'hashtags': [],
 'owner': {'id': '436a1012-3c5d-4c84-879b-a4e434f43230',
           'nickname': 'TNEO',
           'image': 'fc85fbc9-ef4a-4045-905d-bd6fb581689c',
           'donationActive': False,
           'block': False,
           'serviceTags': []},
 'reactions': {'totalCount': 0, 'myReaction': None, 'counters': []},
 'content': {'type': 'simple',
             'text': '',
             'instruments': [{'type': 'share',
                              'ticker': 'AAPL',
                              'lastPrice': 176.02,
                              'currency': 'usd',
                              'image': 'US0378331005.png',
                              'briefName': 'Apple',
                              'dailyYield': None,
                              'relativeDailyYield': 0.0,
                              'price': 175.34,
                              'relativeYield': 0.39}],
             'hashtags': [],
             'profiles': [],
             'images': [],
             'strategies': []},
 'baseTariffCategory': 'unauthorized',
 'isBookmarked': False,
 'status': 'published'}