claude_dev
3fbe6c9be5
- 添加 vnpy_v4.4.0/ 目录到版本控制 - 删除嵌入的 .git 目录避免嵌套仓库 - 添加 README_SANGUO.md 说明来源和用途 - 更新 .gitignore 移除 vnpy_v4.4.0/ 忽略规则 上游代码信息: - 项目: VeighNa 量化交易平台 - 版本: 4.4.0 - 来源: https://github.com/vnpy/vnpy 用途: 版本参考、代码学习、差异对比、离线开发 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
35 lines
681 B
Python
35 lines
681 B
Python
from time import sleep
|
|
from typing import Any
|
|
from vnpy.rpc import RpcClient
|
|
|
|
|
|
class TestClient(RpcClient):
|
|
"""
|
|
Test RpcClient
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
"""
|
|
Constructor
|
|
"""
|
|
super().__init__()
|
|
|
|
def callback(self, topic: str, data: Any) -> None:
|
|
"""
|
|
Realize callable function
|
|
"""
|
|
print(f"client received topic:{topic}, data:{data}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
req_address = "tcp://localhost:2014"
|
|
sub_address = "tcp://localhost:4102"
|
|
|
|
tc = TestClient()
|
|
tc.subscribe_topic("")
|
|
tc.start(req_address, sub_address)
|
|
|
|
while 1:
|
|
print(tc.add(1, 3))
|
|
sleep(2)
|