本公众号所有内容仅为个人量化技术研究、思路分享与案例分析，不构成任何投资建议或股票推荐。金融市场具有较高风险，所有操作决策需建立在独立判断之上。文中提及的任何策略、指标或方法均存在局限性，过往表现不代表未来收益，且可能随市场环境变化而失效。文章仅为技术分享学习使用，不可直接用于实盘。EasyXT项目介绍
EasyXT是基于miniqmt中xtquant的二次开发封装库，旨在简化xtquant的使用，提供更友好的API接口。通过统一的接口设计、智能参数处理和完善的错误处理，让量化交易开发变得更加简单高效。
项目地址: https://github.com/quant-king299/EasyXT
🛠️ 环境准备
系统要求
- 操作系统：Windows 10/11（PowerShell 7）
- Python：3.9+（建议 3.10+），并将 Python 加入 PATH
ptrade/QMT账号获取指导
📱 还没有ptrade/QMT账号的朋友，可以扫码加我微信，全程指导搞定Ptrade/QMT账号！
引言
在前几篇文章中,我们学习了环境搭建、数据获取和数据分析。现在,到了最激动人心的时刻——实战策略代码!
本文将提供4个经过实际测试验证的交易策略信号生成器,每个策略都包含:
所有代码都已经过实际测试验证!
注意: 本文提供的是策略信号生成工具,用于辅助决策,不构成投资建议。
📊 策略1: 双均线交叉策略
1.1 策略原理
双均线交叉是最经典的技术分析策略之一:
优点: 逻辑简单,趋势跟踪效果好缺点: 震荡市容易频繁交易
1.2 信号生成代码
from tqcenter import tqimport pandas as pdclassDualMAStrategy:"""双均线交叉策略 - 信号生成器"""def__init__(self, fast_period=5, slow_period=20):""" Args: fast_period: 快线周期(默认5日) slow_period: 慢线周期(默认20日) """self.fast_period = fast_periodself.slow_period = slow_perioddefgenerate_signals(self, stock_code, start_date, end_date):""" 生成交易信号 Returns: DataFrame: 包含价格、均线和信号的DataFrame """ tq.initialize('dual_ma_strategy.py')# 获取数据 data = tq.get_market_data( field_list=['Close'], stock_list=[stock_code], period='1d', start_time=start_date, end_time=end_date )if'Close'notin data or stock_code notin data['Close']:print(f"[ERROR] 无法获取 {stock_code} 的数据") tq.close()returnNone df = data['Close'][[stock_code]].copy() df.columns = ['Close']# 计算均线 df[f'MA{self.fast_period}'] = df['Close'].rolling(self.fast_period).mean() df[f'MA{self.slow_period}'] = df['Close'].rolling(self.slow_period).mean()# 生成信号 df['Signal_Type'] = 'hold'for i inrange(1, len(df)):# 前一日和当日均线 ma_fast_prev = df[f'MA{self.fast_period}'].iloc[i-1] ma_slow_prev = df[f'MA{self.slow_period}'].iloc[i-1] ma_fast_curr = df[f'MA{self.fast_period}'].iloc[i] ma_slow_curr = df[f'MA{self.slow_period}'].iloc[i]# 跳过无效数据if pd.isna(ma_fast_prev) or pd.isna(ma_slow_prev):continue# 金叉: 买入信号if ma_fast_prev <= ma_slow_prev and ma_fast_curr > ma_slow_curr: df.loc[df.index[i], 'Signal_Type'] = 'buy'# 死叉: 卖出信号elif ma_fast_prev >= ma_slow_prev and ma_fast_curr < ma_slow_curr: df.loc[df.index[i], 'Signal_Type'] = 'sell'# 只返回有信号的行 signals = df[df['Signal_Type'] != 'hold']print(f"\n{stock_code} 双均线策略信号:")print(f"期间: {start_date} - {end_date}")print(f"参数: MA{self.fast_period} / MA{self.slow_period}")print(f"\n信号数量: {len(signals)}")iflen(signals) > 0:print(signals[['Close', f'MA{self.fast_period}', f'MA{self.slow_period}', 'Signal_Type']].tail(10)) tq.close()return df# 使用示例if __name__ == '__main__': strategy = DualMAStrategy(fast_period=5, slow_period=20) df = strategy.generate_signals('600519.SH', '20241101', '20241231')if df isnotNone:# 显示最近的信号 recent_signals = df[df['Signal_Type'] != 'hold'].tail(5)print("\n最近的交易信号:")for date, row in recent_signals.iterrows():print(f" {date}: {row['Signal_Type']} @ {row['Close']:.2f}")
1.3 实际测试结果
600519.SH 双均线策略信号:期间: 20241101 - 20241231参数: MA5 / MA20信号数量: 2 Close MA5 MA20 Signal_Type2024-12-12 1565.80 1537.560 1529.9290 buy2024-12-26 1527.79 1529.012 1533.4025 sell最近的交易信号: 2024-12-12: buy @ 1565.80 2024-12-26: sell @ 1527.79
1.4 参数优化建议
📈 策略2: MACD策略
2.1 策略原理
MACD(Moving Average Convergence Divergence)是趋势跟踪动量指标:
买入信号: MACD线上穿信号线(柱状图由负转正)卖出信号: MACD线下穿信号线(柱状图由正转负)
2.2 信号生成代码
from tqcenter import tqimport pandas as pdclassMACDStrategy:"""MACD策略 - 信号生成器"""def__init__(self, fast_period=12, slow_period=26, signal_period=9):""" Args: fast_period: 快EMA周期(默认12) slow_period: 慢EMA周期(默认26) signal_period: 信号线周期(默认9) """self.fast_period = fast_periodself.slow_period = slow_periodself.signal_period = signal_perioddefcalculate_macd(self, prices):"""计算MACD指标"""# 计算EMA ema_fast = prices.ewm(span=self.fast_period).mean() ema_slow = prices.ewm(span=self.slow_period).mean()# MACD线 macd = ema_fast - ema_slow# 信号线 signal = macd.ewm(span=self.signal_period).mean()# 柱状图 histogram = macd - signalreturn macd, signal, histogramdefgenerate_signals(self, stock_code, start_date, end_date):"""生成交易信号""" tq.initialize('macd_strategy.py')# 获取数据 data = tq.get_market_data( field_list=['Close'], stock_list=[stock_code], period='1d', start_time=start_date, end_time=end_date )if'Close'notin data or stock_code notin data['Close']:print(f"[ERROR] 无法获取 {stock_code} 的数据") tq.close()returnNone df = data['Close'][[stock_code]].copy() df.columns = ['Close']# 计算MACD macd, signal, histogram = self.calculate_macd(df['Close']) df['MACD'] = macd df['Signal'] = signal df['Histogram'] = histogram# 生成信号 df['Signal_Type'] = 'hold'for i inrange(1, len(df)): macd_prev = df['MACD'].iloc[i-1] signal_prev = df['Signal'].iloc[i-1] macd_curr = df['MACD'].iloc[i] signal_curr = df['Signal'].iloc[i]# 跳过无效数据if pd.isna(macd_prev) or pd.isna(signal_prev):continue# 金叉: MACD上穿信号线if macd_prev <= signal_prev and macd_curr > signal_curr: df.loc[df.index[i], 'Signal_Type'] = 'buy'# 死叉: MACD下穿信号线elif macd_prev >= signal_prev and macd_curr < signal_curr: df.loc[df.index[i], 'Signal_Type'] = 'sell'# 只返回有信号的行 signals = df[df['Signal_Type'] != 'hold']print(f"\n{stock_code} MACD策略信号:")print(f"参数: MACD({self.fast_period}, {self.slow_period}, {self.signal_period})")print(f"\n信号数量: {len(signals)}")iflen(signals) > 0:print(signals[['Close', 'MACD', 'Signal', 'Signal_Type']].tail(10)) tq.close()return df# 使用示例if __name__ == '__main__': strategy = MACDStrategy() df = strategy.generate_signals('600519.SH', '20241101', '20241231')if df isnotNone:# 显示最近的信号 recent_signals = df[df['Signal_Type'] != 'hold'].tail(5)print("\n最近的交易信号:")for date, row in recent_signals.iterrows():print(f" {date}: {row['Signal_Type']} @ {row['Close']:.2f}")
2.3 实际测试结果
600519.SH MACD策略信号:参数: MACD(12, 26, 9)信号数量: 4 Close MACD Signal Signal_Type2024-11-04 1548.20 0.324199 0.180110 buy2024-11-12 1577.20 2.900406 3.119230 sell2024-12-09 1518.80 -9.232665 -9.503024 buy2024-12-26 1527.79 -1.571384 -1.370801 sell最近的交易信号: 2024-11-04: buy @ 1548.20 2024-11-12: sell @ 1577.20 2024-12-09: buy @ 1518.80 2024-12-26: sell @ 1527.79
2.4 进阶技巧
# 只在MACD柱状图>0时买入if macd_curr > signal_curr and histogram_curr > 0: signal = 'buy'
📊 策略3: 布林带突破策略
3.1 策略原理
布林带由三条线组成:
买入信号: 价格触及下轨后反弹卖出信号: 价格触及上轨后回落
3.2 信号生成代码
from tqcenter import tqimport pandas as pdimport numpy as npclassBollingerBandsStrategy:"""布林带策略 - 信号生成器"""def__init__(self, period=20, std_dev=2):""" Args: period: 均线周期(默认20) std_dev: 标准差倍数(默认2) """self.period = periodself.std_dev = std_devdefcalculate_bollinger_bands(self, prices):"""计算布林带""" sma = prices.rolling(window=self.period).mean() std = prices.rolling(window=self.period).std() upper = sma + (std * self.std_dev) lower = sma - (std * self.std_dev)return sma, upper, lowerdefgenerate_signals(self, stock_code, start_date, end_date):"""生成交易信号""" tq.initialize('bollinger_strategy.py')# 获取数据 data = tq.get_market_data( field_list=['Close'], stock_list=[stock_code], period='1d', start_time=start_date, end_time=end_date )if'Close'notin data or stock_code notin data['Close']:print(f"[ERROR] 无法获取 {stock_code} 的数据") tq.close()returnNone df = data['Close'][[stock_code]].copy() df.columns = ['Close']# 计算布林带 sma, upper, lower = self.calculate_bollinger_bands(df['Close']) df['BB_Middle'] = sma df['BB_Upper'] = upper df['BB_Lower'] = lower# 生成信号 df['Signal_Type'] = 'hold'for i inrange(self.period, len(df)): price_today = df['Close'].iloc[i] price_prev = df['Close'].iloc[i-1] lower_today = df['BB_Lower'].iloc[i] upper_today = df['BB_Upper'].iloc[i] lower_prev = df['BB_Lower'].iloc[i-1] middle_today = df['BB_Middle'].iloc[i]# 跳过无效数据if pd.isna(lower_today) or pd.isna(upper_today):continue# 买入: 前一日触及下轨,当日反弹if price_prev <= lower_prev and price_today > lower_today: df.loc[df.index[i], 'Signal_Type'] = 'buy'# 卖出: 价格触及上轨elif price_today >= upper_today: df.loc[df.index[i], 'Signal_Type'] = 'sell'# 止损: 价格跌破中轨elif price_today < middle_today: df.loc[df.index[i], 'Signal_Type'] = 'sell' signals = df[df['Signal_Type'] != 'hold']print(f"\n{stock_code} 布林带策略信号:")print(f"参数: 周期={self.period}, 标准差={self.std_dev}")print(f"\n信号数量: {len(signals)}")iflen(signals) > 0:print(signals[['Close', 'BB_Upper', 'BB_Middle', 'BB_Lower', 'Signal_Type']].tail(10)) tq.close()return df# 使用示例if __name__ == '__main__': strategy = BollingerBandsStrategy(period=20, std_dev=2) df = strategy.generate_signals('600519.SH', '20241101', '20241231')if df isnotNone:# 显示最近的信号 recent_signals = df[df['Signal_Type'] != 'hold'].tail(5)print("\n最近的交易信号:")for date, row in recent_signals.iterrows():print(f" {date}: {row['Signal_Type']} @ {row['Close']:.2f}")
🎯 策略4: 多因子选股策略
4.1 策略原理
综合多个因子进行选股:
- 价格动量因子
- 波动率因子
- 成交量因子
综合得分 → 选出得分最高的股票
4.2 信号生成代码
from tqcenter import tqimport pandas as pdimport numpy as npclassMultiFactorStrategy:"""多因子选股策略"""def__init__(self, lookback_days=20, top_n=5):""" Args: lookback_days: 回看天数 top_n: 选出前N只股票 """self.lookback_days = lookback_daysself.top_n = top_ndefcalculate_factors(self, stock_code, start_date, end_date):"""计算单个股票的因子得分"""# 获取价格数据 price_data = tq.get_market_data( field_list=['Close', 'Volume'], stock_list=[stock_code], period='1d', start_time=start_date, end_time=end_date ) factors = {}# 因子1: 价格动量if'Close'in price_data and stock_code in price_data['Close']: prices = price_data['Close'][stock_code]iflen(prices) >= self.lookback_days: start_price = prices.iloc[-self.lookback_days] end_price = prices.iloc[-1] momentum = (end_price - start_price) / start_price * 100 factors['价格动量'] = momentum# 因子2: 波动率(负向)if'Close'in price_data and stock_code in price_data['Close']: prices = price_data['Close'][stock_code]iflen(prices) >= self.lookback_days: returns = prices.tail(self.lookback_days).pct_change() volatility = returns.std() factors['波动率'] = -volatility * 100# 负号# 因子3: 成交量增长if'Volume'in price_data and stock_code in price_data['Volume']: volumes = price_data['Volume'][stock_code]iflen(volumes) >= self.lookback_days: recent = volumes.tail(self.lookback_days) first_half = recent.iloc[:self.lookback_days//2].mean() second_half = recent.iloc[self.lookback_days//2:].mean()if first_half > 0: volume_growth = (second_half - first_half) / first_half * 100 factors['成交量增长'] = volume_growthreturn factorsdefselect_stocks(self, stock_pool, start_date, end_date):"""选股"""print(f"\n多因子选股策略")print(f"股票池: {len(stock_pool)}只")print(f"选出: {self.top_n}只") tq.initialize('multi_factor_strategy.py') scores = {}for stock in stock_pool:try: factors = self.calculate_factors(stock, start_date, end_date)if factors:# 综合得分(权重可调整) total_score = ( factors.get('价格动量', 0) * 0.4 + factors.get('波动率', 0) * 0.3 + factors.get('成交量增长', 0) * 0.3 ) scores[stock] = {'score': total_score,'factors': factors }except Exception as e:pass# 排序并选出top N ranked = sorted(scores.items(), key=lambda x: x[1]['score'], reverse=True) top_stocks = ranked[:self.top_n]print(f"\n选股结果:")for i, (stock, data) inenumerate(top_stocks, 1):print(f"\n{i}. {stock}")print(f" 综合得分: {data['score']:.2f}")for factor, value in data['factors'].items():print(f" {factor}: {value:.2f}") tq.close()return top_stocks# 使用示例if __name__ == '__main__':# 获取股票池 tq.initialize('get_stocks.py') stock_pool = tq.get_stock_list('1')[:30] # 前30只 tq.close()# 运行选股 strategy = MultiFactorStrategy(lookback_days=20, top_n=5) selected = strategy.select_stocks(stock_pool, '20241101', '20241231')
📊 策略对比总结
🎓 实战建议
1. 策略选择
2. 参数优化
# 网格搜索最优参数fast_periods = [5, 10, 15]slow_periods = [20, 30, 40]for fast in fast_periods:for slow in slow_periods: strategy = DualMAStrategy(fast, slow) signals = strategy.generate_signals('600519.SH', '20241101', '20241231')# 记录结果
3. 风险控制
# 信号过滤deffilter_signals(signals, min_interval=5):"""过滤信号,确保两次信号间隔足够长""" filtered = [] last_signal_date = Nonefor date, signal in signals:if last_signal_date isNoneor \ (date - last_signal_date).days >= min_interval: filtered.append((date, signal)) last_signal_date = datereturn filtered
4. 组合策略
# 多个策略组合确认defgenerate_combined_signal(stock_code):"""组合策略信号"""# 策略1: 双均线 ma_strategy = DualMAStrategy() ma_signals = ma_strategy.generate_signals(stock_code, '20241101', '20241231')# 策略2: MACD macd_strategy = MACDStrategy() macd_signals = macd_strategy.generate_signals(stock_code, '20241101', '20241231')# 只有当多个策略同时给出信号时才采纳# ...return combined_signal
📚 总结
本文提供了4个经过实际测试验证的交易策略信号生成器:
所有策略都:
测试环境: Windows 10, Python 3.11.7, 通达信金融终端V7.72
免责声明: 本文仅供学习研究使用,不构成投资建议。投资有风险,入市需谨慎。
通达信论坛->宽客社区 (https://sns.tdx.com.cn)通达信量化帮助文档 (https://help.tdx.com.cn/quant)通达信下载中心 (https://www.tdx.com.cn/soft.html)📱 关注我们
欢迎扫码持续关注公众号，会持续分享
🔍 公众号名称: 王者quant📚 分享内容: 量化交易、Python编程、投资策略🎯 更新频率: 持续更新，干货满满
通过公众号您可以获得：
本教程仅供学习参考，实际交易请谨慎操作！
