cyb50-quant/live_trade.py

#!/usr/bin/env python3
"""
实盘交易执行脚本
连接券商API（QMT/Ptrade）执行实际交易
"""

import json
import logging
from datetime import datetime
from typing import Dict, List
import sys

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('live_trade.log', encoding='utf-8'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)


class TradeExecutor:
    """交易执行器"""
    
    def __init__(self, broker_api=None):
        self.broker_api = broker_api  # 券商API实例
        self.simulation_mode = True  # 默认模拟模式
        
    def connect(self, config: Dict):
        """连接券商API"""
        # 这里接入QMT或Ptrade的API
        # from xtquant import xtdata, xttype, xttrader
        logger.info("连接交易API...")
        
        if self.simulation_mode:
            logger.info("当前为模拟交易模式，不会执行真实交易")
        else:
            logger.info("已连接真实交易接口")
            
    def buy(self, code: str, name: str, price: float, amount: float, 
            strategy: str) -> bool:
        """买入"""
        shares = int(amount / price / 100) * 100  # 整手
        if shares < 100:
            logger.warning(f"{name}({code}) 资金不足，无法买入")
            return False
            
        logger.info(f"[BUY] {name}({code}): {shares}股 @ {price:.2f}, 策略:{strategy}")
        
        if not self.simulation_mode:
            # 执行真实交易
            # order_id = self.broker_api.buy(code, shares, price)
            pass
            
        return True
        
    def sell(self, code: str, name: str, price: float, shares: int, 
             reason: str, strategy: str) -> bool:
        """卖出"""
        logger.info(f"[SELL] {name}({code}): {shares}股 @ {price:.2f}, 原因:{reason}, 策略:{strategy}")
        
        if not self.simulation_mode:
            # 执行真实交易
            # order_id = self.broker_api.sell(code, shares, price)
            pass
            
        return True
        
    def get_positions(self) -> Dict:
        """获取当前持仓"""
        # 从券商API获取持仓
        return {}
        
    def get_account(self) -> Dict:
        """获取账户信息"""
        return {
            'total_asset': 1000000,
            'available_cash': 100000,
            'market_value': 900000
        }


class LiveTrader:
    """实盘交易主类"""
    
    def __init__(self, config_path: str = "config.json"):
        self.config = self.load_config(config_path)
        self.executor = TradeExecutor()
        self.executor.simulation_mode = True  # 安全第一，默认模拟
        
    def load_config(self, path: str) -> Dict:
        """加载配置"""
        try:
            with open(path, 'r', encoding='utf-8') as f:
                return json.load(f)
        except Exception as e:
            logger.error(f"加载配置失败: {e}")
            return {}
    
    def run_daily(self):
        """每日执行"""
        logger.info("=" * 60)
        logger.info("开始实盘交易")
        logger.info(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        logger.info("=" * 60)
        
        # 1. 获取账户状态
        account = self.executor.get_account()
        logger.info(f"账户总资产: {account['total_asset']:,.0f}")
        logger.info(f"可用资金: {account['available_cash']:,.0f}")
        
        # 2. 获取当前持仓
        positions = self.executor.get_positions()
        
        # 3. 检查风险
        # risk_check()
        
        # 4. 执行策略信号
        # 这里调用quant_system.py中的策略逻辑
        
        logger.info("实盘交易执行完成")
        
    def enable_live_trading(self):
        """启用真实交易（谨慎！）"""
        confirm = input("⚠️ 即将启用真实交易模式，输入 'YES' 确认: ")
        if confirm == "YES":
            self.executor.simulation_mode = False
            logger.info("已切换到真实交易模式")
        else:
            logger.info("保持模拟模式")


def main():
    """主函数"""
    trader = LiveTrader()
    
    print("=" * 60)
    print("量化交易系统 - 实盘交易")
    print("=" * 60)
    print("\n选择操作:")
    print("1. 运行每日策略 (模拟模式)")
    print("2. 启用真实交易 (⚠️ 危险)")
    print("3. 查看账户信息")
    print("4. 退出")
    
    choice = input("\n请输入选项 (1-4): ").strip()
    
    if choice == "1":
        trader.run_daily()
    elif choice == "2":
        trader.enable_live_trading()
    elif choice == "3":
        account = trader.executor.get_account()
        print(f"\n总资产: {account['total_asset']:,.0f}元")
        print(f"可用资金: {account['available_cash']:,.0f}元")
        print(f"持仓市值: {account['market_value']:,.0f}元")
    elif choice == "4":
        print("退出")
    else:
        print("无效选项")


if __name__ == "__main__":
    main()