real_trader/examples/example_trade.py

import requests

# 服务器地址
URL = "http://trader.biggerfish.tech:9527/yu"
# 策略名称常量
STRATEGY = "港股ETF网格"

def buy(code: str, price: float, amount: int, strategy_name: str = STRATEGY) -> dict:
    """买入股票
    
    Args:
        code: 股票代码，例如 "601988.SH"
        price: 买入价格
        amount: 买入数量
        strategy_name: 策略名称
    """
    data = {
        "code": code,
        "price": str(price),
        "amount": str(amount),
        "strategy_name": strategy_name
    }
    response = requests.post(f"{URL}/buy", json=data)
    return response.json()

def sell(code: str, price: float, amount: int, strategy_name: str = STRATEGY) -> dict:
    """卖出股票
    
    Args:
        code: 股票代码，例如 "601988.SH"
        price: 卖出价格
        amount: 卖出数量
        strategy_name: 策略名称
    """
    data = {
        "code": code,
        "price": str(price),
        "amount": str(amount),
        "strategy_name": strategy_name
    }
    response = requests.post(f"{URL}/sell", json=data)
    return response.json()

def get_positions(strategy_name: str = STRATEGY) -> dict:
    """获取持仓信息
    
    Args:
        strategy_name: 策略名称，默认为STRATEGY常量
    """
    params = {}
    if strategy_name:
        params["strategy_name"] = strategy_name
    response = requests.get(f"{URL}/positions", params=params)
    return response.json()

def get_target_positions(strategy_name: str = STRATEGY) -> dict:
    """获取目标持仓信息（仅实盘模式有效）
    
    Args:
        strategy_name: 策略名称，默认为STRATEGY常量
    """
    params = {"target": "true"}
    if strategy_name:
        params["strategy_name"] = strategy_name
    response = requests.get(f"{URL}/positions", params=params)
    return response.json()

def clear_strategy(strategy_name: str = STRATEGY) -> dict:
    """清除策略持仓数据
    
    Args:
        strategy_name: 策略名称
    """
    response = requests.delete(f"{URL}/clear/{strategy_name}")
    return response.json()

def get_balance() -> dict:
    """获取账户余额（仅实盘模式）
    
    Returns:
        字典形式的账户余额信息
    """
    response = requests.get(f"{URL}/balance")
    return response.json()

def get_today_trades() -> dict:
    """获取今日成交记录（仅实盘模式）
    
    Returns:
        字典形式的今日成交记录
    """
    response = requests.get(f"{URL}/todaytrades")
    return response.json()

def get_today_orders() -> dict:
    """获取今日委托记录（仅实盘模式）
    
    Returns:
        字典形式的今日委托记录
    """
    response = requests.get(f"{URL}/todayentrust")
    return response.json()

def cancel_order(entrust_no: str) -> dict:
    """取消订单
    
    Args:
        entrust_no: 委托编号
        
    Returns:
        取消结果
    """
    response = requests.delete(f"{URL}/cancel/{entrust_no}")
    return response.json()

def get_order_status() -> dict:
    """获取订单状态
    
    Returns:
        字典形式的订单状态信息
    """
    response = requests.get(f"{URL}/order_status")
    return response.json()

def get_strategy_targets(strategy_name: str = STRATEGY) -> dict:
    """获取策略目标持仓（仅实盘模式）
    
    Args:
        strategy_name: 策略名称，默认为STRATEGY常量
        
    Returns:
        字典形式的策略目标持仓信息
    """
    params = {}
    if strategy_name:
        params["strategy_name"] = strategy_name
    response = requests.get(f"{URL}/strategy_targets", params=params)
    return response.json()

def check_health() -> str:
    """健康检查
    
    Returns:
        健康状态
    """
    response = requests.get(f"{URL}/healthcheck")
    return response.text


if __name__ == "__main__":
    # 示例：健康检查
    health = check_health()
    print("健康状态:", health)
    
    # 示例：查询默认策略的持仓信息
    positions = get_positions()
    print("当前持仓:", positions)
    # 模拟输出:
    #  {'data': [], 'success': True}
    
    # 示例：查询账户余额（仅实盘模式）
    balance = get_balance()
    print("账户余额:", balance)
    
    # 示例：买入中国银行1000股，价格3.45
    result = buy("601988.SH", 3.45, 3000)
    print("买入结果:", result)
    # 模拟输出:
    #{'data': {'message': '模拟买入 - 代码: 601988.SH, 价格: 3.45, 数量: 3000', 'order_id': 'simulation'}, 'success': True}
    
    # 示例：获取订单状态
    order_status = get_order_status()
    print("订单状态:", order_status)
    
    # 示例：卖出中国银行1000股，价格3.48
    result = sell("601988.SH", 3.48, 1000)
    print("卖出结果:", result)
    # 模拟输出:
    #  {'data': {'message': '模拟卖出 - 代码: 601988.SH, 价格: 3.48, 数量: 1000', 'order_id': 'simulation'}, 'success': True}
    
    # 示例：获取今日成交记录
    trades = get_today_trades()
    print("今日成交:", trades)
    
    # 示例：获取今日委托记录
    entrusts = get_today_orders()
    print("今日委托:", entrusts)

    # 示例：再次查询持仓变化
    positions = get_positions()
    print("交易后持仓:", positions)
    # 模拟输出:
    # {'data': [{'601988.SH': {'closeable_amount': 2000, 'total_amount': 2000}}], 'success': True}
    
    # 示例：获取目标持仓
    target_positions = get_target_positions()
    print("目标持仓:", target_positions)
    
    # 示例：获取策略目标持仓
    strategy_targets = get_strategy_targets()
    print("策略目标持仓:", strategy_targets)
    
    # 示例：取消订单（需要真实的委托编号）
    # cancel_result = cancel_order("123456")
    # print("取消订单结果:", cancel_result)
    
    # 示例：清除策略持仓数据
    result = clear_strategy()
    print("清除策略持仓结果:", result)
    # 模拟输出:
    # {'message': "成功清除策略 '追梦投资港股ETF' 的持仓数据", 'success': True}
    
    # 示例：清除后查询持仓
    positions = get_positions()
    print("清除后持仓:", positions)
    # 模拟输出:
    # {'data': [], 'success': True}