实战指南：基于qstock与EasyXT构建你的首个自动化量化交易系统

# qstock数据获取模块
try:
    import qstock as qs
    QSTOCK_AVAILABLE = True
    print("✅ qstock数据模块加载成功")
except ImportError as e:
    print(f"❌ qstock模块导入失败: {e}")
    QSTOCK_AVAILABLE = False

# EasyXT交易执行模块
try:
    from easy_xt.api import EasyXT
    EASYXT_AVAILABLE = True
    print("✅ EasyXT交易模块加载成功")
except ImportError as e:
    print(f"❌ EasyXT模块导入失败: {e}")
    EASYXT_AVAILABLE = False

class QStockEasyXTIntegration:
    def __init__(self):
        # 数据存储结构
        self.data_cache = {}
        self.signal_history = []
        self.trade_history = []
        
        # 系统运行状态
        self.is_trading_enabled = False
        self.is_monitoring = False
        
        # 按步骤初始化各模块
        self.init_data_module()      # 初始化qstock数据模块
        self.init_trading_module()   # 初始化EasyXT交易模块

def get_multi_source_data(self, symbol: str, period: int = 60):
    """使用qstock获取指定标的多源数据"""
    data_dict = {}
    # 计算起止日期
    end_date = datetime.now().strftime('%Y%m%d')
    start_date = (datetime.now() - timedelta(days=period)).strftime('%Y%m%d')
    
    # 1. 获取历史K线数据
    kline_data = qs.get_data(symbol, start=start_date, end=end_date)
    data_dict['kline'] = self.clean_kline_data(kline_data)
    
    # 2. 获取实时行情快照
    realtime_data = qs.get_realtime([symbol])
    data_dict['realtime'] = realtime_data
    
    # 3. 获取资金流向数据
    fund_flow = qs.get_fund_flow([symbol])
    data_dict['fund_flow'] = fund_flow
    
    # 4. 获取最新财务数据
    financial_data = qs.get_financial_data(symbol)
    data_dict['financial'] = financial_data
    
    # 5. 获取相关新闻舆情
    news_data = qs.get_news(symbol)
    data_dict['news'] = news_data
    
    return data_dict

def clean_kline_data(self, data: pd.DataFrame) -> pd.DataFrame:
    """清洗与标准化K线数据DataFrame"""
    # 列名标准化映射
    column_mapping = {
        'Open': 'open', 'High': 'high', 'Low': 'low', 
        'Close': 'close', 'Volume': 'volume'
    }
    data.rename(columns=column_mapping, inplace=True)
    
    # 基础数据清洗：删除空值及零成交量记录
    data = data.dropna()
    data = data[data['volume'] > 0]
    
    # 数值类型强制转换，避免后续计算错误
    for col in ['open', 'high', 'low', 'close', 'volume']:
        data[col] = pd.to_numeric(data[col], errors='coerce')
    
    return data

def calculate_technical_indicators(self, data: pd.DataFrame):
    """在DataFrame上计算并添加多种技术指标"""
    # 移动平均线
    data['MA5'] = data['close'].rolling(window=5).mean()
    data['MA20'] = data['close'].rolling(window=20).mean()
    
    # MACD指标
    data['EMA12'] = data['close'].ewm(span=12).mean()
    data['EMA26'] = data['close'].ewm(span=26).mean()
    data['MACD'] = data['EMA12'] - data['EMA26']
    data['MACD_signal'] = data['MACD'].ewm(span=9).mean()
    data['MACD_hist'] = data['MACD'] - data['MACD_signal']
    
    # RSI相对强弱指标
    delta = data['close'].diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
    data['RSI'] = 100 - (100 / (1 + gain / loss))
    
    # 布林带
    data['BB_middle'] = data['close'].rolling(window=20).mean()
    bb_std = data['close'].rolling(window=20).std()
    data['BB_upper'] = data['BB_middle'] + (bb_std * 2)
    data['BB_lower'] = data['BB_middle'] - (bb_std * 2)
    
    return data

def generate_trading_signals(self, symbol: str, data: pd.DataFrame):
    """基于多策略分析，生成综合交易信号"""
    signal_strength = 0  # 信号强度累积值，正为看多，负为看空
    signal_reasons = []  # 记录触发信号的具体原因
    
    # 策略1: 趋势跟踪（基于均线排列）
    trend_signals = self._trend_following_strategy(data)
    signal_strength += trend_signals['strength']
    signal_reasons.extend(trend_signals.get('reasons', []))
    
    # 策略2: 均值回归（基于RSI等超买超卖指标）
    mean_reversion_signals = self._mean_reversion_strategy(data)
    signal_strength += mean_reversion_signals['strength']
    signal_reasons.extend(mean_reversion_signals.get('reasons', []))
    
    # 策略3: 动量策略（基于价格变化速率）
    momentum_signals = self._momentum_strategy(data)
    signal_strength += momentum_signals['strength']
    signal_reasons.extend(momentum_signals.get('reasons', []))
    
    # 策略4: 成交量确认（价量齐升/跌）
    volume_signals = self._volume_confirmation_strategy(data)
    signal_strength += volume_signals['strength']
    signal_reasons.extend(volume_signals.get('reasons', []))
    
    # 策略5: K线形态识别（如头肩顶、W底等）
    pattern_signals = self._pattern_recognition_strategy(data)
    signal_strength += pattern_signals['strength']
    signal_reasons.extend(pattern_signals.get('reasons', []))
    
    # 综合评估，将强度转换为交易方向和置信度
    signal_type = 'BUY' if signal_strength > 0 else 'SELL'
    confidence = min(95, max(0, 50 + abs(signal_strength) * 10))  # 置信度范围0-95%
    
    return {
        'symbol': symbol,
        'signal_type': signal_type,
        'strength': signal_strength,
        'confidence': confidence,
        'reasons': signal_reasons,
        'price': data['close'].iloc[-1],
        'timestamp': datetime.now()
    }

def execute_trading_signal(self, signal: Dict):
    """执行交易信号的全流程"""
    # 1. 获取最新的账户资金信息
    account_info = self.get_account_info()
    
    # 2. 获取标的当前持仓情况
    position_info = self.get_position_info(signal['symbol'])
    
    # 3. 执行多层次风险检查
    risk_check = self.risk_management_check(signal, account_info, position_info)
    if not risk_check['passed']:
        return {'status': 'rejected', 'message': risk_check['reason']}
    
    # 4. 基于风控规则和账户状态，计算本次建议交易的数量
    quantity = self.calculate_trade_quantity(signal, account_info, position_info)
    if quantity <= 0:
        return {'status': 'skipped', 'message': '计算出的交易数量为0'}
    
    # 5. 调用EasyXT接口执行交易
    if signal['signal_type'] == 'BUY':
        result = self.execute_buy_order(signal['symbol'], quantity, signal['price'])
    else:  # SELL
        result = self.execute_sell_order(signal['symbol'], quantity, signal['price'])
    
    return result

def get_account_info(self):
    """获取交易账户的资产概况"""
    # 调用EasyXT底层接口
    account_info = self.trader.get_account_asset(TRADING_CONFIG['account_id'])
    # 结构化返回信息
    return {
        'total_asset': account_info.get('total_asset', 0),
        'cash': account_info.get('cash', 0),
        'market_value': account_info.get('market_value', 0),
        'profit_loss': account_info.get('profit_loss', 0)
    }

def get_position_info(self, symbol: str):
    """获取指定标的的持仓详情"""
    positions = self.trader.get_positions(TRADING_CONFIG['account_id'], symbol)
    if not positions.empty:
        position = positions.iloc[0]
        return {
            'volume': position.get('volume', 0),           # 持仓数量
            'can_use_volume': position.get('can_use_volume', 0), # 可卖数量
            'cost_price': position.get('cost_price', 0),   # 成本价
            'market_value': position.get('market_value', 0)# 持仓市值
        }
    return {'volume': 0, 'can_use_volume': 0, 'cost_price': 0, 'market_value': 0}

def risk_management_check(self, signal, account_info, position_info):
    """执行交易前的风险管理检查清单"""
    total_asset = account_info.get('total_asset', 100000)
    current_position_value = position_info.get('market_value', 0)
    
    # 检查1: 总仓位上限控制
    max_position_value = total_asset * TRADING_CONFIG['max_position_ratio']
    if signal['signal_type'] == 'BUY':
        planned_trade_value = signal['price'] * 100  # 按最小交易单位估算
        if current_position_value + planned_trade_value > max_position_value:
            return {'passed': False, 'reason': '超过最大仓位限制'}
    
    # 检查2: 单标的风险暴露控制
    single_stock_max = total_asset * TRADING_CONFIG['single_stock_ratio']
    if current_position_value > single_stock_max:
        return {'passed': False, 'reason': '超过单股最大仓位'}
    
    # 检查3: 持仓标的的浮动止损检查
    if position_info.get('volume', 0) > 0:
        cost_price = position_info.get('cost_price', 0)
        current_price = signal['price']
        if cost_price > 0:
            loss_ratio = (cost_price - current_price) / cost_price
            if loss_ratio > TRADING_CONFIG['stop_loss_ratio']:
                return {'passed': False, 'reason': '触发浮动止损线'}
    
    # 检查4: 信号质量过滤器
    if signal['confidence'] < STRATEGY_CONFIG['signal_threshold']:
        return {'passed': False, 'reason': '信号置信度不足'}
    
    return {'passed': True, 'reason': '所有风险检查均通过'}

def calculate_trade_quantity(self, signal, account_info, position_info):
    """基于风控规则和账户状态，动态计算建议交易数量"""
    if signal['signal_type'] == 'BUY':
        # 买入逻辑：根据可用资金的一定比例和信号强度计算
        available_cash = account_info.get('cash', 0)
        # 基础分配比例，可根据信号强度进一步调整
        trade_cash_ratio = 0.3
        planned_cash = available_cash * trade_cash_ratio
        
        # 考虑买入手续费（粗略估算）
        price_with_fee = signal['price'] * 1.001
        # 计算数量并取整到100股（A股最小交易单位）
        quantity = int(planned_cash / price_with_fee) // 100 * 100
        return max(100, quantity)  # 确保至少1手
    else:
        # 卖出逻辑：根据可卖数量和信号强度决定卖出比例
        can_sell_volume = position_info.get('can_use_volume', 0)
        if can_sell_volume > 0:
            # 信号强度绝对值越大，卖出比例越高（但不超过50%）
            sell_ratio = min(0.5, abs(signal['strength']))
            quantity = int(can_sell_volume * sell_ratio) // 100 * 100
            return max(100, min(quantity, can_sell_volume))  # 确保在1手到可卖数量之间
        return 0

def start_real_time_monitoring(self):
    """启动后台实时监控线程"""
    self.is_monitoring = True
    # 创建并启动监控线程（设置为守护线程，主程序退出时自动结束）
    monitor_thread = threading.Thread(target=self._monitoring_loop, daemon=True)
    monitor_thread.start()
    print("✅ 实时监控系统已启动")

def _monitoring_loop(self):
    """监控主循环逻辑"""
    while self.is_monitoring:
        print(f"\n🔄 实时监控更新 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        all_signals = []
        
        # 遍历不同分类的股票池（如核心股、成长股、价值股等）
        for category, stocks in STOCK_POOL.items():
            print(f"📊 监控 {category}板块...")
            for stock in stocks[:2]:  # 示例中每类只监控前两只，以控制频率
                # 获取该股票的多维度数据
                data_dict = self.get_multi_source_data(stock, period=30)
                if 'kline' in data_dict and not data_dict['kline'].empty:
                    # 计算指标并生成信号
                    kline_data = self.calculate_technical_indicators(data_dict['kline'])
                    signals = self.generate_trading_signals(stock, kline_data)
                    all_signals.extend([signals]) if isinstance(signals, dict) else all_signals.extend(signals)
                    # 打印关键信息
                    latest = kline_data.iloc[-1]
                    print(f"  {stock}: 价格 {latest['close']:.2f}, RSI {latest.get('RSI', 50):.1f}")
        
        # 筛选并提示高质量信号
        high_quality_signals = [s for s in all_signals if s['confidence'] >= STRATEGY_CONFIG['signal_threshold']]
        for signal in high_quality_signals:
            print(f"🔥 高质量信号: {signal['symbol']} {signal['signal_type']} (置信度: {signal['confidence']:.1f}%)")
        
        # 显示当前账户状态摘要
        self._display_account_status()
        
        # 等待下一个监控周期
        time.sleep(STRATEGY_CONFIG['update_interval'])

def _display_account_status(self):
    """在监控面板中显示账户关键信息"""
    account_info = self.get_account_info()
    print(f"\n💼 账户状态摘要:")
    print(f"  总资产: {account_info.get('total_asset', 0):,.2f}")
    print(f"  可用资金: {account_info.get('cash', 0):,.2f}")
    print(f"  持仓市值: {account_info.get('market_value', 0):,.2f}")
    print(f"  浮动盈亏: {account_info.get('profit_loss', 0):,.2f}")

def run_backtest(self, symbol: str, start_date: str, end_date: str):
    """在指定历史时期运行策略回测"""
    # 1. 获取历史数据
    historical_data = qs.get_data(symbol, start=start_date, end=end_date)
    
    # 2. 数据清洗与指标计算
    historical_data = self.clean_kline_data(historical_data)
    historical_data = self.calculate_technical_indicators(historical_data)
    
    # 3. 模拟交易：遍历历史每一天，应用策略逻辑
    backtest_results = self._simulate_trading(symbol, historical_data)
    
    # 4. 计算绩效指标
    performance_metrics = self._calculate_performance_metrics(backtest_results)
    
    # 5. 生成图文报告并保存
    self._generate_backtest_report(symbol, backtest_results, performance_metrics)
    
    return {
        'symbol': symbol,
        'period': f"{start_date} 至 {end_date}",
        'trades': backtest_results,
        'performance': performance_metrics
    }

def _calculate_performance_metrics(self, trades):
    """基于模拟交易记录计算详细的绩效指标"""
    if not trades:
        return {}
    
    # 基础统计
    total_trades = len(trades)
    buy_trades = [t for t in trades if t['action'] == 'BUY']
    sell_trades = [t for t in trades if t['action'] == 'SELL']
    
    # 资金曲线与收益计算（假设初始资金10万）
    initial_capital = 100000
    # 简化计算：根据最后的交易记录获取最终总资产
    final_value = trades[-1]['total_value'] if trades else initial_capital
    total_return = (final_value - initial_capital) / initial_capital
    
    # 交易对分析（买入后对应卖出为一个完整回合）
    trade_pairs = []
    for i in range(min(len(buy_trades), len(sell_trades))):
        buy = buy_trades[i]
        sell = sell_trades[i]
        if sell['date'] > buy['date']:  # 确保卖出在买入之后
            profit = (sell['price'] - buy['price']) * buy['quantity']
            profit_rate = profit / (buy['price'] * buy['quantity'])
            trade_pairs.append({'profit': profit, 'profit_rate': profit_rate})
    
    # 关键指标计算
    winning_trades = [p for p in trade_pairs if p['profit'] > 0]
    win_rate = len(winning_trades) / len(trade_pairs) if trade_pairs else 0
    avg_profit = np.mean([p['profit'] for p in trade_pairs]) if trade_pairs else 0
    
    # 计算最大回撤（简化版，需基于每日资产序列精细计算）
    # 这里仅为示意，实际实现更复杂
    max_drawdown = -2.15  # 示例值
    
    return {
        'total_trades': total_trades,
        'trade_pairs': len(trade_pairs),
        'total_return': total_return,
        'annualized_return': total_return * (252/len(trades)) if trades else 0, # 年化
        'win_rate': win_rate,
        'avg_profit': avg_profit,
        'max_drawdown': max_drawdown,
        'sharpe_ratio': 1.85  # 示例值，需基于日收益率序列计算
    }

def create_visualization(self, symbol: str, data: pd.DataFrame, signals: List[Dict]):
    """创建包含价格、指标和信号标记的综合分析图表"""
    import matplotlib.pyplot as plt
    fig, axes = plt.subplots(3, 1, figsize=(15, 12))
    fig.suptitle(f'{symbol} qstock+EasyXT 量化分析', fontsize=16, fontweight='bold')
    
    # 子图1: 价格与移动平均线
    ax1 = axes[0]
    ax1.plot(data.index, data['close'], label='收盘价', linewidth=2, color='black')
    ax1.plot(data.index, data['MA5'], label='MA5', alpha=0.7, linestyle='--')
    ax1.plot(data.index, data['MA20'], label='MA20', alpha=0.7, linestyle='-.')
    # 标记买卖信号点
    for signal in signals:
        if signal['signal_type'] == 'BUY':
            ax1.scatter(data.index[-1], signal['price'], color='red', marker='^', s=100, label='买入信号' if signal == signals[0] else "")
        else:
            ax1.scatter(data.index[-1], signal['price'], color='green', marker='v', s=100, label='卖出信号' if signal == signals[0] else "")
    ax1.legend()
    ax1.set_ylabel('价格')
    ax1.grid(True, alpha=0.3)
    
    # 子图2: RSI指标
    ax2 = axes[1]
    ax2.plot(data.index, data['RSI'], label='RSI', color='purple', linewidth=1.5)
    ax2.axhline(y=70, color='r', linestyle='--', alpha=0.5, label='超买线(70)')
    ax2.axhline(y=30, color='g', linestyle='--', alpha=0.5, label='超卖线(30)')
    ax2.fill_between(data.index, 30, 70, color='gray', alpha=0.1)
    ax2.legend()
    ax2.set_ylabel('RSI')
    ax2.set_ylim(0, 100)
    ax2.grid(True, alpha=0.3)
    
    # 子图3: MACD指标
    ax3 = axes[2]
    ax3.plot(data.index, data['MACD'], label='MACD', color='blue', linewidth=1.5)
    ax3.plot(data.index, data['MACD_signal'], label='Signal', color='red', linewidth=1)
    ax3.bar(data.index, data['MACD_hist'], label='Histogram', alpha=0.3, color=['g' if x>=0 else 'r' for x in data['MACD_hist']])
    ax3.legend()
    ax3.set_ylabel('MACD')
    ax3.set_xlabel('日期')
    ax3.grid(True, alpha=0.3)
    
    plt.tight_layout()
    
    # 保存图表文件
    chart_file = f"reports/{symbol}_analysis_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
    plt.savefig(chart_file, dpi=300, bbox_inches='tight')
    plt.close(fig)  # 关闭图形，释放内存
    
    return chart_file