Client SDK¶

EasyTHS 提供了官方的 Python Client SDK（TradeClient），用于与服务端进行通信，执行各种交易操作。

安装¶

Client SDK 包含在 easyths 包中。根据您的使用场景，可以选择以下两种安装方式：

仅安装客户端 SDK（推荐用于远程调用）¶

如果您只需要使用 TradeClient 连接到已运行的服务端，可以仅安装基础包：

# 使用 pip 安装（推荐）
pip install easyths

# 或使用 uv
uv add easyths

客户端模式仅依赖：

httpx - HTTP 客户端
pydantic - 数据验证

安装完整服务端（包含客户端）¶

如果您需要在本地运行完整的服务端（包括自动化交易功能），需要安装服务端版本：

# 使用 pip 安装（推荐）
pip install easyths[server]

# 或使用 uv
uv add easyths[server]

完整服务端包含：

所有客户端依赖
FastAPI 服务端
pywinauto（Windows GUI 自动化）
其他服务端依赖（OCR、图像处理等）

注意：完整服务端仅支持 Windows 系统。客户端 SDK 可以在任何系统上运行。

快速开始¶

基本用法¶

from easyths import TradeClient

# 创建客户端
client = TradeClient(
    host="127.0.0.1",
    port=7648,
    api_key="your-api-key"  # 如果配置了 API Key
)

# 健康检查
health = client.health_check()
print(health)

# 使用完毕后关闭连接
client.close()

使用上下文管理器（推荐）¶

from easyths import TradeClient

# 使用 with 语句自动管理连接
with TradeClient(host="127.0.0.1", port=7648, api_key="your-api-key") as client:
    health = client.health_check()
    print(health)
# 连接会自动关闭

初始化参数¶

参数	类型	默认值	说明
host	str	"127.0.0.1"	服务端主机地址
port	int	7648	服务端端口
api_key	str	""	API 密钥（用于身份验证）
timeout	float	30.0	请求超时时间（秒）
scheme	str	"http"	协议方案（http/https）

系统管理¶

健康检查¶

health = client.health_check()
# 返回: {"success": True, "message": "系统运行正常", "data": {...}}

获取系统状态¶

status = client.get_system_status()
# 返回: {"success": True, "data": {"automator": {...}, "plugins": {...}}}

获取系统信息¶

info = client.get_system_info()
# 返回: {"success": True, "data": {"name": "...", "version": "..."}}

获取队列统计¶

stats = client.get_queue_stats()
# 返回: {"success": True, "data": {"queued_count": 0, ...}}

获取可用操作列表¶

ops = client.list_operations()
# 返回: {"success": True, "data": {"operations": {...}, "count": 7}}

交易操作¶

买入股票¶

result = client.buy(
    stock_code="600000",  # 股票代码
    price=10.50,          # 买入价格
    quantity=100          # 买入数量（100的倍数）
)

# 检查结果
if result["success"]:
    data = result["data"]
    print(f"买入成功: {data['message']}")
else:
    error = result["message"]
    print(f"买入失败: {error}")

卖出股票¶

result = client.sell(
    stock_code="600000",
    price=11.00,
    quantity=100
)

if result["success"]:
    print("卖出成功")

撤销委托单¶

# 撤销所有委托
result = client.cancel_order()

# 撤销指定股票的委托
result = client.cancel_order(stock_code="600000")

# 只撤销买单
result = client.cancel_order(cancel_type="buy")

# 只撤销卖单
result = client.cancel_order(cancel_type="sell")

条件买入¶

设置条件买入单，当股价达到目标价格时自动触发买入。

result = client.condition_buy(
    stock_code="600000",      # 股票代码
    target_price=10.50,       # 目标触发价格
    quantity=100,             # 买入数量（100的倍数）
    expire_days=30            # 有效期（可选1/3/5/10/20/30，默认30）
)

if result["success"]:
    data = result["data"]
    print(f"条件买入设置成功: {data['message']}")
else:
    error = result["message"]
    print(f"条件买入设置失败: {error}")

止盈止损¶

为持仓股票设置止盈止损策略，当价格达到止盈或止损条件时自动触发卖出。

result = client.stop_loss_profit(
    stock_code="600000",         # 股票代码
    stop_loss_percent=3.0,       # 止损百分比（如3表示3%）
    stop_profit_percent=5.0,     # 止盈百分比（如5表示5%）
    quantity=100,                # 卖出数量（可选，不指定则使用全部可用持仓）
    expire_days=30               # 有效期（可选1/3/5/10/20/30，默认30）
)

if result["success"]:
    data = result["data"]
    print(f"止盈止损设置成功: {data['message']}")
else:
    error = result["message"]
    print(f"止盈止损设置失败: {error}")

注意：止盈百分比必须大于止损百分比。quantity 参数建议指定，因为受 T+1 限制，当天买入的股票如果不指定数量无法设置止盈止损。

购买国债逆回购¶

result = client.reverse_repo_buy(
    market="上海",        # 交易市场：上海/深圳
    time_range="1天期",   # 回购期限：1天期/2天期/3天期/4天期/7天期
    amount=10000          # 出借金额（1000的倍数）
)

if result["success"]:
    message = result["data"]["message"]
    print(f"购买成功: {message}")
else:
    error = result["message"]
    print(f"购买失败: {error}")

删除条件单¶

删除指定的条件单。

# 删除所有条件单
result = client.cancel_condition_orders()

# 删除指定股票的条件单
result = client.cancel_condition_orders(stock_code="600000")

# 只删除买入条件单
result = client.cancel_condition_orders(order_type="买入")

# 删除指定股票的买入条件单
result = client.cancel_condition_orders(
    stock_code="600000",
    order_type="买入"
)

if result["success"]:
    message = result["data"]["message"]
    print(f"删除成功: {message}")

查询操作¶

查询持仓¶

result = client.query_holdings(
    return_type="json"  # str/json/dict/markdown
)

if result["success"]:
    holdings = result["data"]["holdings"]
    for position in holdings:
        print(f"{position['股票代码']}: {position['持仓数量']}股")

查询资金¶

result = client.query_funds()

if result["success"]:
    funds = result["data"]
    print(f"总资产: {funds['总资产']}")
    print(f"可用金额: {funds['可用金额']}")

查询委托单¶

# 查询所有委托
result = client.query_orders(return_type="json")

# 查询指定股票的委托
result = client.query_orders(
    stock_code="600000",
    return_type="json"
)

if result["success"]:
    orders = result["data"]["orders"]
    for order in orders:
        print(f"{order['股票代码']}: {order['委托数量']}股 @ {order['委托价格']}")

查询历史成交¶

result = client.query_historical_commission(return_type="json")

if result["success"]:
    commissions = result["data"]
    print(commissions)

查询国债逆回购年化利率¶

result = client.query_reverse_repo()

if result["success"]:
    rates = result["data"]["reverse_repo_interest"]
    for item in rates:
        print(f"{item['市场类型']} - {item['时间类型']}: {item['年化利率']}")

查询条件单¶

查询未触发的条件单信息。

result = client.query_condition_orders(
    return_type="json"  # str/json/dict/markdown
)

if result["success"]:
    orders = result["data"]["condition_orders"]
    print(orders)

通用操作方法¶

执行操作¶

# 执行自定义操作
operation_id = client.execute_operation(
    operation_name="buy",
    params={
        "stock_code": "600000",
        "price": 10.50,
        "quantity": 100
    },
    priority=5  # 优先级 0-10，数字越大优先级越高
)
print(f"操作ID: {operation_id}")

获取操作状态¶

status = client.get_operation_status(operation_id)
print(status)
# 返回状态: queued/running/success/failed

获取操作结果¶

# 阻塞等待直到操作完成
result = client.get_operation_result(
    operation_id=operation_id
)

if result["success"]:
    print("操作成功:", result["data"])

取消操作¶

# 取消排队中的操作
success = client.cancel_operation(operation_id)

异常处理¶

SDK 提供了 TradeClientError 异常类，用于处理各种错误：

from easyths import TradeClient, TradeClientError

try:
    with TradeClient(host="127.0.0.1", port=7648) as client:
        result = client.buy("600000", 10.50, 100)
        if result["success"]:
            print("买入成功")

except TradeClientError as e:
    print(f"交易失败: {e}")
    if e.status_code:
        print(f"状态码: {e.status_code}")

常见错误状态码：

连接失败：无法连接到服务端
401：认证失败（API Key 错误）
408：操作超时
500：服务端内部错误

完整示例¶

简单交易脚本¶

from easyths import TradeClient, TradeClientError

def simple_trade():
    """简单的交易示例"""
    with TradeClient(
        host="127.0.0.1",
        port=7648,
        api_key="your-api-key"
    ) as client:
        # 检查系统健康
        health = client.health_check()
        if not health["success"]:
            print("系统异常")
            return

        # 查询资金
        funds = client.query_funds()
        if funds["success"]:
            available = funds["data"]["可用金额"]
            print(f"可用资金: {available}")

        # 买入股票
        result = client.buy("600000", 10.50, 100)
        if result["success"]:
            print("买入成功")
        else:
            print(f"买入失败: {result['message']}")

if __name__ == "__main__":
    try:
        simple_trade()
    except TradeClientError as e:
        print(f"错误: {e}")

异步操作示例¶

from easyths import TradeClient, TradeClientError
import time

def async_trade_example():
    """异步提交多个操作"""
    with TradeClient(host="127.0.0.1", port=7648) as client:
        operation_ids = []

        # 提交多个买入操作
        stocks = [("600000", 10.50), ("600036", 35.00), ("000001", 12.00)]
        for code, price in stocks:
            op_id = client.execute_operation(
                "buy",
                {"stock_code": code, "price": price, "quantity": 100},
                priority=5
            )
            operation_ids.append(op_id)
            print(f"已提交买入 {code}，操作ID: {op_id}")

        # 等待所有操作完成
        results = []
        for op_id in operation_ids:
            result = client.get_operation_result(op_id)
            results.append(result)

        # 处理结果
        for result in results:
            if result["success"]:
                data = result["data"]
                print(f"操作成功: {data.get('message', 'N/A')}")
            else:
                print(f"操作失败: {result['message']}")

API 参考¶

TradeClient 类¶

class TradeClient:
    def __init__(
        self,
        host: str = "127.0.0.1",
        port: int = 7648,
        api_key: str = "",
        timeout: float = 30.0,
        scheme: str = "http"
    ): ...

    # 系统管理
    def health_check(self) -> dict: ...
    def get_system_status(self) -> dict: ...
    def get_system_info(self) -> dict: ...
    def get_queue_stats(self) -> dict: ...
    def list_operations(self) -> dict: ...

    # 通用操作
    def execute_operation(self, operation_name: str, params: dict, priority: int = 0) -> str: ...
    def get_operation_status(self, operation_id: str) -> dict: ...
    def get_operation_result(self, operation_id: str, timeout: float = None) -> dict: ...
    def cancel_operation(self, operation_id: str) -> bool: ...

    # 交易操作
    def buy(self, stock_code: str, price: float, quantity: int, timeout: float = None) -> dict: ...
    def sell(self, stock_code: str, price: float, quantity: int, timeout: float = None) -> dict: ...
    def cancel_order(self, stock_code: str = None, cancel_type: str = "all", timeout: float = None) -> dict: ...
    def condition_buy(self, stock_code: str, target_price: float, quantity: int, expire_days: int = 30, timeout: float = None) -> dict: ...
    def stop_loss_profit(self, stock_code: str, stop_loss_percent: float, stop_profit_percent: float, quantity: int = None, expire_days: int = 30, timeout: float = None) -> dict: ...
    def query_condition_orders(self, return_type: str = "json", timeout: float = None) -> dict: ...
    def cancel_condition_orders(self, stock_code: str = None, order_type: str = None, timeout: float = None) -> dict: ...
    def reverse_repo_buy(self, market: str, time_range: str, amount: int, timeout: float = None) -> dict: ...

    # 查询操作
    def query_holdings(self, return_type: str = "json", timeout: float = None) -> dict: ...
    def query_funds(self, timeout: float = None) -> dict: ...
    def query_orders(self, stock_code: str = None, return_type: str = "json", timeout: float = None) -> dict: ...
    def query_historical_commission(self, return_type: str = "json", timeout: float = None) -> dict: ...
    def query_reverse_repo(self, timeout: float = None) -> dict: ...

    # 连接管理
    def close(self): ...
    def __enter__(self): ...
    def __exit__(self, exc_type, exc_val, exc_tb): ...

交易操作返回格式¶

所有交易操作（buy, sell, condition_buy 等）返回 OperationResult 格式：

{
    "success": bool,        # 业务操作是否成功
    "data": {...},          # 业务数据
    "message": str | None,  # 错误信息或成功消息
    "timestamp": str        # 操作时间（ISO 8601 格式）
}

示例：

result = client.buy("600000", 10.50, 100)
# {
#     "success": True,
#     "data": {
#         "stock_code": "600000",
#         "price": "10.50",
#         "quantity": 100,
#         "operation": "buy",
#         "success": True,
#         "message": "成功提交600000的买入委托"
#     },
#     "message": None,
#     "timestamp": "2025-12-26T10:30:00.123456"
# }

系统接口返回格式¶

系统管理接口（health_check, get_system_status 等）返回 APIResponse 格式：

{
    "success": bool,      # 操作是否成功
    "message": str,       # 响应消息
    "data": Any,          # 响应数据
    "timestamp": str      # 响应时间戳（ISO 8601 格式）
}

TradeClientError 异常类¶

客户端 SDK 提供了专用的异常类 TradeClientError，用于处理客户端级别的错误：

class TradeClientError(Exception):
    """客户端异常"""

    def __init__(self, message: str, status_code: Optional[int] = None):
        """
        Args:
            message: 错误消息
            status_code: HTTP 状态码（可选）
        """

属性：

属性	类型	说明
message	str	错误消息描述
status_code	int \| None	HTTP 状态码（如果有）

使用示例：

from easyths import TradeClient, TradeClientError

try:
    with TradeClient(host="127.0.0.1", port=7648) as client:
        result = client.buy("600000", 10.50, 100)
except TradeClientError as e:
    print(f"错误消息: {e}")
    print(f"状态码: {e.status_code}")

常见异常场景：

场景	status_code	说明
连接失败	None	无法连接到服务端，请检查服务端是否启动
认证失败	401	API Key 错误或未提供
操作超时	408	操作执行时间超过设定的超时时间
服务端错误	500	服务端内部错误
HTTP 错误	其他	HTTP 请求失败，对应相应的 HTTP 状态码