备注：回测结果只作为历史数据分析，不作为投资依据。
回测对象：A股所有主板股票
回测周期：20250101~20251231
数据文件格式如下：
图片
打开今日头条查看图片详情
图片
打开今日头条查看图片详情
全市场回测报告交易总次数: 12571
胜率: 39.91%
平均单笔收益: 0.99%
平均持仓天数: 8.2天
盈亏比: 1 : 2.16
图片
打开今日头条查看图片详情
通达信主图图片
打开今日头条查看图片详情
MA3:=MA(CLOSE,3);MA5:MA(CLOSE,5),DOTLINE;MA10:MA(CLOSE,10),DOTLINE;MA20:MA(CLOSE,20),DOTLINE;MA30:MA(CLOSE,30),DOTLINE;MA60:MA(CLOSE,60),LINETHICK4;MA120:MA(CLOSE,120),LINETHICK2;MA240:MA(CLOSE,240),LINETHICK2;主板 := CODELIKE('00') OR CODELIKE('60');双创 := CODELIKE('30') OR CODELIKE('68');北交所 := CODELIKE('4') OR CODELIKE('8');涨停幅度 := IF(主板, 0.1, IF(双创, 0.2, IF(北交所, 0.3, 100)));涨停 := CLOSE = HIGH AND CLOSE >= ZTPRICE(REF(CLOSE, 1), 涨停幅度);STICKLINE(涨停,HIGH,LOW,0,0),COLORFF00FF;STICKLINE(涨停,CLOSE,OPEN,2,0),COLORFF00FF;STICKLINE(涨停,CLOSE,OPEN,2,0),COLORFF00FF;近一年高点:REF(HHV(HIGH,240),1),COLORYELLOW,DOTLINE;TYP:=(HIGH+LOW+CLOSE)/3;能量:=(TYP-MA(TYP,89))*1000/(15*AVEDEV(TYP,89));控盘度:=WINNER(C)*100;MAX_MA := MAX(MAX(MAX(MA5, MA10), MA20), MA30);MIN_MA := MIN(MIN(MIN(MA5, MA10), MA20), MA30);粘合度 := (MAX_MA - MIN_MA) / MIN_MA * 100;粘合启动 := 粘合度 <= 5 AND CLOSE > MAX_MA AND VOL / MA(VOL, 5) > 1.3 AND (HIGH / LOW > 1.05 OR CLOSE/REF(CLOSE,1)>=1.03);主升浪:=能量>=100 AND (COUNT(CROSS(控盘度,50),3)+COUNT(粘合启动,3))>0;DRAWICON(主升浪,LOW*0.99,9);出货:=CROSS(MA5,MA3);DRAWICON(出货 AND REF(BARSLAST(出货)>BARSLAST(主升浪),1),HIGH*1.01,6);{基本面信息}所属板块:=STRCAT('【行业板块】：',STRCAT(' [行业]--',HYBLOCK));风格板块:=STRCAT('【概念板块】：',STRCAT(' [概念]--',GNBLOCK));总股本:=STRCAT(STRCAT('总股本:',CON2STR(TOTALCAPITAL/1000000,2)),'亿');流通盘:=STRCAT(STRCAT(' 流通盘:',CON2STR(CAPITAL/1000000,2)),'亿');全流通:=STRCAT(STRCAT(' 流通比:',CON2STR(CAPITAL/TOTALCAPITAL*100,2)),'％');市值:=STRCAT(STRCAT(' 流通市值:',CON2STR((CAPITAL/1000000)*C,2)),'亿');股本市值:=STRCAT('【股本市值】：',STRCAT(STRCAT(STRCAT(总股本,流通盘),全流通),市值));市盈率:=STRCAT('市盈率:',CON2STR(C/FINANCE(33),2));净资收益率:=STRCAT(' 净资收益率:',CON2STR(FINANCE(33)/FINANCE(34)*100,2));主营利润率:=STRCAT(' 主营利润率:',CON2STR(FINANCE(23)/FINANCE(20)*100,2));当前业绩:=STRCAT('【当前业绩】：',STRCAT(STRCAT(市盈率,净资收益率),主营利润率));DRAWTEXT_FIX(ISLASTBAR,0.005,0.02,0,所属板块),COLORYELLOW;DRAWTEXT_FIX(ISLASTBAR ,0.005,0.07,0,风格板块),COLORYELLOW;DRAWTEXT_FIX(ISLASTBAR,0.005,0.12,0,股本市值),COLORMAGENTA;DRAWTEXT_FIX(ISLASTBAR,0.005,0.17,0,当前业绩),COLORMAGENTA;两年最低:=LLV(L,500);两年涨幅点数:=((C-两年最低)/两年最低)*100;DRAWTEXT_FIX(ISLASTBAR,0.85,0.1,0,'相对于两年最低价涨幅(%):'),COLORWHITE;DRAWNUMBER_FIX(ISLASTBAR,0.93,0.1,0,两年涨幅点数),COLORYELLOW;成交额:=AMOUNT/100000000;一年最低:=LLV(L,250);涨幅点数:=((C-一年最低)/一年最低)*100; DRAWTEXT_FIX(ISLASTBAR,0.85,0.15,0,'成交额(亿):'),COLORWHITE;DRAWNUMBER_FIX(ISLASTBAR,0.9,0.15,0,成交额),COLORYELLOW;DRAWTEXT_FIX(ISLASTBAR,0.85,0.20,0,'相对于一年最低价涨幅(%):'),COLORWHITE;DRAWNUMBER_FIX(ISLASTBAR,0.93,0.20,0,涨幅点数),COLORYELLOW;振幅:=((H - L) / L) * 100;DRAWTEXT_FIX(ISLASTBAR,0.85,0.25,0,'今天振幅(%) :'),COLORWHITE;DRAWNUMBER_FIX(ISLASTBAR,0.9,0.25,0,振幅),COLORYELLOW;DRAWTEXT_FIX(ISLASTBAR,0.85,0.30,0,'MA5斜率 :'),COLORWHITE;DRAWNUMBER_FIX(ISLASTBAR,0.9,0.30,0,MA5斜率),COLORYELLOW;通达信选股图片
打开今日头条查看图片详情
图片
打开今日头条查看图片详情
备注：出现信号的标的比较多，建议加入自己其他逻辑一起。
主板 := CODELIKE('00') OR CODELIKE('60');MA3:=MA(CLOSE,3);MA5:=MA(CLOSE,5);MA10:=MA(CLOSE,10);MA20 := MA(CLOSE, 20);MA30 := MA(CLOSE, 30);MA60 := MA(CLOSE, 60);TYP:=(HIGH+LOW+CLOSE)/3;能量:=(TYP-MA(TYP,89))*1000/(15*AVEDEV(TYP,89));控盘度:=WINNER(C)*100;MAX_MA := MAX(MAX(MAX(MA5, MA10), MA20), MA30);MIN_MA := MIN(MIN(MIN(MA5, MA10), MA20), MA30);粘合度 := (MAX_MA - MIN_MA) / MIN_MA * 100;粘合启动 := 粘合度 <= 5 AND CLOSE > MAX_MA AND VOL / MA(VOL, 5) > 1.3 AND (HIGH / LOW > 1.05 OR CLOSE/REF(CLOSE,1)>=1.03);主升浪:主板 AND 能量>=100 AND (COUNT(CROSS(控盘度,50),3)+COUNT(粘合启动,3))>0;Python回测代码import osimport globimport datetimeimport warningsimport numpy as npimport pandas as pdimport matplotlib.pyplot as pltimport seaborn as snsfrom concurrent.futures import ProcessPoolExecutor# --- 1. 全局配置 ---DATA_PATH = r'E:\quant\history'encoding_type = 'utf-8'warnings.filterwarnings('ignore')# 设置绘图的中文字体 (解决方框乱码问题)plt.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei', 'Arial Unicode MS']plt.rcParams['axes.unicode_minus'] = False# 数学常量PI = 3.141592653589793def process_one_stock(file_path): try: df = pd.read_csv( file_path, encoding=encoding_type, usecols=['交易时间', '开盘价', '收盘价', '最高价', '最低价', '停牌']) # 过滤掉停牌日 (假设 1 代表停牌) df = df[df['停牌'] != 1] # 过滤新股 if len(df) < 60: return None rename_dict = {'交易时间': 'date', '开盘价': 'open', '收盘价': 'close', '最高价': 'high', '最低价': 'low'} df.rename(columns=rename_dict, inplace=True) df['date'] = pd.to_datetime(df['date']) df.sort_values('date', inplace=True) df.reset_index(drop=True, inplace=True) # --- 3. 策略逻辑：主升浪策略 (按通达信公式修改) --- # 计算基础均线 df['ma3'] = df['close'].rolling(window=3, min_periods=3).mean() df['ma5'] = df['close'].rolling(window=5, min_periods=5).mean() df['ma10'] = df['close'].rolling(window=10, min_periods=10).mean() df['ma20'] = df['close'].rolling(window=20, min_periods=20).mean() df['ma30'] = df['close'].rolling(window=30, min_periods=30).mean() df['ma60'] = df['close'].rolling(window=60, min_periods=60).mean() # 1. 计算均线斜率（角度） df['ma5_prev'] = df['ma5'].shift(1) df['ma10_prev'] = df['ma10'].shift(1) # 处理除零问题 df['ma5_ratio'] = np.where(df['ma5_prev'] != 0, (df['ma5'] / df['ma5_prev'] - 1) * 100, 0) df['ma10_ratio'] = np.where(df['ma10_prev'] != 0, (df['ma10'] / df['ma10_prev'] - 1) * 100, 0) df['ma5_slop'] = np.arctan(df['ma5_ratio']) * 180 / PI df['ma10_slop'] = np.arctan(df['ma10_ratio']) * 180 / PI # 2. 计算破新高 df['hhv_240'] = df['close'].rolling(window=240, min_periods=1).max().shift(1) df['break_high'] = df['high'] >= df['hhv_240'] # 3. 计算能量指标 (TYP和能量) df['typ'] = (df['high'] + df['low'] + df['close']) / 3 df['ma_typ_89'] = df['typ'].rolling(window=89, min_periods=89).mean() df['avedev_typ_89'] = df['typ'].rolling(window=89, min_periods=89).apply( lambda x: np.mean(np.abs(x - np.mean(x))), raw=True ) # 处理除零问题 df['energy'] = np.where(df['avedev_typ_89'] != 0, (df['typ'] - df['ma_typ_89']) * 1000 / (15 * df['avedev_typ_89']), 0) # 4. 判断是否为主板股票 (从文件名提取代码判断) code = os.path.basename(file_path).replace('.csv', '') df['is_main_board'] = (code.startswith('00') or code.startswith('60')) # 5. 计算控盘度 (WINNER函数近似：收盘价处的获利盘比例) # 注：Python中无直接WINNER函数，用滚动20日的价格区间近似 df['high_20'] = df['high'].rolling(window=20, min_periods=20).max() df['low_20'] = df['low'].rolling(window=20, min_periods=20).min() df['control_degree'] = np.where(df['high_20'] != df['low_20'], (df['close'] - df['low_20']) / (df['high_20'] - df['low_20']) * 100, 0) # 6. 计算粘合度和粘合启动 # 计算MAX_MA和MIN_MA df['max_ma'] = df[['ma5', 'ma10', 'ma20', 'ma30']].max(axis=1) df['min_ma'] = df[['ma5', 'ma10', 'ma20', 'ma30']].min(axis=1) # 粘合度 df['cohesion'] = np.where(df['min_ma'] != 0, (df['max_ma'] - df['min_ma']) / df['min_ma'] * 100, 0) # 成交量相关 df['vol_ma5'] = df['成交量'].rolling(window=5, min_periods=5).mean() if '成交量' in df.columns else 0 df['vol_ratio'] = np.where(df['vol_ma5'] != 0, df['成交量'] / df['vol_ma5'], 0) if '成交量' in df.columns else 0 # 粘合启动条件 df['cohesion_start'] = ( (df['cohesion'] <= 5) & (df['close'] > df['max_ma']) & (df['vol_ratio'] > 1.3) & ((df['high'] / df['low'] > 1.05) | (df['close'] / df['close'].shift(1) >= 1.03)) ) # 7. 计算CROSS(控盘度,50)和COUNT统计 # 控盘度上穿50 df['cross_control_50'] = (df['control_degree'].shift(1) < 50) & (df['control_degree'] >= 50) # 过去3天内CROSS(控盘度,50)的次数 df['count_cross_control'] = df['cross_control_50'].rolling(window=3, min_periods=1).sum() # 过去3天内粘合启动的次数 df['count_cohesion_start'] = df['cohesion_start'].rolling(window=3, min_periods=1).sum() # --- 核心：主升浪买入条件 (严格按通达信公式) --- df['main_rise'] = ( df['is_main_board'] & (df['energy'] >= 100) & ((df['count_cross_control'] + df['count_cohesion_start']) > 0) ) # --- 核心：出货卖出条件 (严格按通达信公式) --- # CROSS(MA5,MA3)：MA5上穿MA3 (代表短期走弱) df['sell_signal_ship'] = (df['ma5'].shift(1) < df['ma3'].shift(1)) & (df['ma5'] >= df['ma3']) # 赋值最终的买卖信号 df['buy_signal'] = df['main_rise'] df['sell_signal'] = df['sell_signal_ship'] # 4. 计算实际持有期收益（根据卖出信号） trades = [] in_position = False buy_price = 0 buy_date = None buy_slop = 0 for i in range(len(df)): # 买入信号 if df.iloc[i]['buy_signal'] and not in_position: in_position = True buy_price = df.iloc[i]['close'] buy_date = df.iloc[i]['date'] buy_slop = df.iloc[i]['ma5_slop'] # 卖出信号（持仓中且出现卖出条件） elif in_position and df.iloc[i]['sell_signal']: sell_price = df.iloc[i]['close'] sell_date = df.iloc[i]['date'] hold_days = (sell_date - buy_date).days # 计算收益率 if buy_price != 0: ret = sell_price / buy_price - 1 trades.append({ 'date': buy_date, 'sell_date': sell_date, 'hold_days': hold_days, 'buy_price': buy_price, 'sell_price': sell_price, 'ret': ret, 'ma5_slop': buy_slop }) in_position = False # 处理最后一条数据仍持仓的情况 elif in_position and i == len(df) - 1: sell_price = df.iloc[i]['close'] sell_date = df.iloc[i]['date'] hold_days = (sell_date - buy_date).days if buy_price != 0: ret = sell_price / buy_price - 1 trades.append({ 'date': buy_date, 'sell_date': sell_date, 'hold_days': hold_days, 'buy_price': buy_price, 'sell_price': sell_price, 'ret': ret, 'ma5_slop': buy_slop }) if not trades: return None # 转换为DataFrame trades_df = pd.DataFrame(trades) # 加上股票代码方便追溯 trades_df['code'] = code # 重命名列以匹配原有结构（ret替代ret_5d，ma5_slop替代dif） trades_df.rename(columns={'ret': 'ret_5d', 'ma5_slop': 'dif'}, inplace=True) return trades_df[['date', 'code', 'buy_price', 'dif', 'ret_5d', 'hold_days', 'sell_date']] except Exception as e: # print(f'Error: {file_path} - {e}') return Nonedef run_backtest_and_plot(): print(f' 开始回测，读取路径: {DATA_PATH}') csv_files = glob.glob(os.path.join(DATA_PATH, '*.csv')) print(f' 扫描到 {len(csv_files)} 个文件，正在全速运算中 (CPU多核并行)...') start_t = datetime.datetime.now() # 并行处理 results = [] with ProcessPoolExecutor() as executor: for res in executor.map(process_one_stock, csv_files): if res is not None: results.append(res) print(f'✅ 计算完成！耗时: {datetime.datetime.now() - start_t}') if not results: print('❌ 悲剧了：没有找到任何符合条件的交易！请检查数据或放宽策略条件。') return # 合并结果 df_all = pd.concat(results, ignore_index=True) df_all.sort_values('date', inplace=True) # --- 统计数据 --- total_trades = len(df_all) win_rate = (df_all['ret_5d'] > 0).mean() avg_ret = df_all['ret_5d'].mean() avg_hold_days = df_all['hold_days'].mean() # 盈亏比估算 avg_win = df_all[df_all['ret_5d'] > 0]['ret_5d'].mean() avg_loss = df_all[df_all['ret_5d'] < 0]['ret_5d'].mean() wl_ratio = abs(avg_win / avg_loss) if avg_loss != 0 else 0 print('=' * 40) print(f' 【主升浪策略】全市场回测报告') print(f'交易总次数: {total_trades}') print(f'胜率: {win_rate:.2%}') print(f'平均单笔收益: {avg_ret:.2%}') print(f'平均持仓天数: {avg_hold_days:.1f}天') print(f'盈亏比: 1 : {wl_ratio:.2f}') print('=' * 40) # 添加样本量检查，避免在数据不足时绘图出错 if len(df_all) < 2: print('⚠️ 交易记录太少，无法生成图表') return # --- 自动画图 (头条号专用素材) --- plot_charts(df_all, win_rate)def plot_charts(df, win_rate): ''' 生成两张精美图表 ''' sns.set_theme(style='whitegrid', rc={'font.sans-serif': ['SimHei', 'Microsoft YaHei']}) fig, axes = plt.subplots(1, 2, figsize=(16, 7)) # 图1：收益分布直方图 # 颜色逻辑：胜率>50%用红，否则用绿，增加视觉冲击力 main_color = '#e74c3c' if win_rate > 0.5 else '#27ae60' # 注意：这里改了 bins 数量，让分布看起来更平滑 sns.histplot(data=df, x='ret_5d', bins=100, kde=True, color=main_color, ax=axes[0], alpha=0.6) axes[0].axvline(0, color='black', linestyle='--', linewidth=2) axes[0].set_title(f'A股全市场回测：主升浪策略收益分布 (胜率{win_rate:.1%})', fontsize=15, fontweight='bold') axes[0].set_xlabel('收益率 (大于0为赚钱)', fontsize=12) axes[0].set_ylabel('触发次数', fontsize=12) # 图2：资金曲线 (模拟复利) # 简单的单利叠加模拟 df['cumsum_ret'] = df['ret_5d'].cumsum() sns.lineplot(data=df, x='date', y='cumsum_ret', color='#2980b9', linewidth=1.5, ax=axes[1]) # 填充颜色 axes[1].fill_between(df['date'], df['cumsum_ret'], 0, color='#2980b9', alpha=0.1) axes[1].set_title('策略累计收益趋势 (样本数N={})'.format(len(df)), fontsize=15, fontweight='bold') axes[1].set_ylabel('累计净值增长 (单利)', fontsize=12) plt.tight_layout() print('️ 图表已生成，请在弹出的窗口中保存图片！') plt.show()if __name__ == '__main__': run_backtest_and_plot()结语授人以鱼不如授之以渔，希望大家看我的文章能学到知识，通达信是个好工具，学会之后复盘效率能提高十倍以上。如果我的文章对您有帮助，别忘了点赞加关注哦。
图片
本站仅提供存储服务，所有内容均由用户发布，如发现有害或侵权内容，请点击举报。七星策略提示：文章来自网络，不代表本站观点。
