yufuquant SDK

$ pip install quantsyssdk

from quantsyssdk.clients import RESTAPIClient

base_url = "https://yufuquant.cc/api/v1"  # 系统后端接口地址
auth_token = "xxxxx"  # 认证令牌
robot_id = 1  # 机器人 id
rest_api_client = RESTAPIClient(base_url=base_url, auth_token=auth_token)

# 获取机器人配置
await rest_api_client.get_robot_config(robot_id)

# 更新机器人的资产信息
data = {
    "total_balance": 5.8
}
await rest_api_client.patch_robot_asset_record(robot_id, data)

# 发送 ping
await rest_api_client.post_robot_ping(robot_id)

import logging
from pprint import pprint
from quantsyssdk.clients import RESTAPIClient

# 开启日志
logger = logging.getLogger("quantsyssdk")
logger.addHandler(logging.StreamHandler())
logger.setLevel(logging.DEBUG)


async def main():
    http_client = RESTAPIClient(
        base_url="https://yufuquant.cc/api/v1",
        auth_token="8g2e470579ba14ea69000859eba6c421b69ff95d",
    )

    result = await http_client.get_robot_config(robot_id=1)
    pprint(result)

    result = await http_client.post_robot_ping(robot_id=1)
    pprint(result)

    result = await http_client.patch_robot_asset_record(
        robot_id=1, data={"total_balance": 10000}
    )
    pprint(result)


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

from quantsyssdk.clients import WebsocketAPIClient

uri = "wss://yufuquant.cc/ws/v1/streams/"  # 系统后端接口地址
auth_token = "xxxxx"  # 认证令牌
topics = ["robot#1.ping", "robot#1.log"]  # 订阅的话题
ws_api_client = WebsocketAPIClient(uri=uri)

# 认证
await ws_api_client.auth(auth_token)

# 订阅话题
await ws_api_client.sub(topics)

# 取消话题订阅
await ws_api_client.unsub(topics)

# 发送机器人 ping
await ws_api_client.robot_ping()

# 发送机器人日志
await ws_api_client.robot_log()

import logging
from quantsyssdk.clients import WebsocketAPIClient

# 开启日志
logger = logging.getLogger("quantsyssdk")
logger.addHandler(logging.StreamHandler())
logger.setLevel(logging.DEBUG)


async def main():
    ws_api_client = WebsocketAPIClient(uri="wss://yufuquant.cc/ws/v1/streams/")

    await ws_api_client.auth("8d2e470575ba04ea69000859eba6c421a69ff95c")
    await ws_api_client.sub(topics=["robot#1.log"])
    while True:
        await ws_api_client.robot_ping()
        await ws_api_client.robot_log("Test robot log...", level="INFO")
        await asyncio.sleep(1)


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())