#!/usr/bin/env python3 """ A股K线指标回测引擎 1. 对每只个股的历史指标触发点进行回测，计算次日上涨概率 2. 判断市场趋势（单边下行识别） 3. 生成交易信号（买入/卖出/观望 + 止损/目标价） """ import pandas as pd import numpy as np import json import os import time import requests from datetime import datetime, timedelta from concurrent.futures import ThreadPoolExecutor, as_completed from collections import defaultdict import warnings warnings.filterwarnings('ignore') import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) WORK_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data") RESULT_FILE = os.path.join(WORK_DIR, "scan_results_filtered.json") OUTPUT_FILE = os.path.join(WORK_DIR, "backtest_results.json") session = requests.Session() session.verify = False session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36', 'Referer': 'https://finance.eastmoney.com/' }) # ===== 复用扫描器的指标计算函数 ===== def calc_ma(series, period): return series.rolling(window=period, min_periods=period).mean() def calc_ema(series, period): return series.ewm(span=period, adjust=False).mean() def calc_macd(close, fast=12, slow=26, signal=9): ema_fast = calc_ema(close, fast) ema_slow = calc_ema(close, slow) dif = ema_fast - ema_slow dea = calc_ema(dif, signal) macd_bar = 2 * (dif - dea) return dif, dea, macd_bar # ===== 量化多因子指标 ===== def calc_rsi(close, period=14): """RSI相对强弱指标""" delta = close.diff() gain = delta.where(delta > 0, 0).rolling(window=period, min_periods=period).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=period, min_periods=period).mean() rs = gain / loss.replace(0, np.nan) rsi = 100 - (100 / (1 + rs)) return rsi.fillna(50) def calc_bollinger(close, period=20, num_std=2): """布林带""" ma = close.rolling(window=period, min_periods=period).mean() std = close.rolling(window=period, min_periods=period).std() upper = ma + num_std * std lower = ma - num_std * std width = (upper - lower) / ma.replace(0, np.nan) # 价格在布林带中的位置 (0=下轨, 1=上轨) pct_b = (close - lower) / (upper - lower).replace(0, np.nan) return upper, lower, width, pct_b def calc_atr(high, low, close, period=14): """ATR平均真实波幅""" tr1 = high - low tr2 = (high - close.shift(1)).abs() tr3 = (low - close.shift(1)).abs() tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1) atr = tr.rolling(window=period, min_periods=period).mean() return atr def calc_adx(high, low, close, period=14): """ADX趋势强度指标""" plus_dm = high.diff() minus_dm = -low.diff() plus_dm = plus_dm.where((plus_dm > minus_dm) & (plus_dm > 0), 0) minus_dm = minus_dm.where((minus_dm > plus_dm) & (minus_dm > 0), 0) tr1 = high - low tr2 = (high - close.shift(1)).abs() tr3 = (low - close.shift(1)).abs() tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1) atr = tr.rolling(window=period, min_periods=period).mean() plus_di = 100 * (plus_dm.rolling(window=period, min_periods=period).mean() / atr.replace(0, np.nan)) minus_di = 100 * (minus_dm.rolling(window=period, min_periods=period).mean() / atr.replace(0, np.nan)) dx = 100 * (plus_di - minus_di).abs() / (plus_di + minus_di).replace(0, np.nan) adx = dx.rolling(window=period, min_periods=period).mean() return adx.fillna(0), plus_di.fillna(0), minus_di.fillna(0) def calc_volume_ratio(volume, period=5): """量比：当前成交量与近期平均成交量之比""" vol_ma = volume.rolling(window=period, min_periods=period).mean() return volume / vol_ma.replace(0, np.nan) def get_stock_data(code, days=500, max_retries=3): """获取个股日K线数据（扩展历史范围用于回测）""" if code.startswith(('00', '30')): secid = f"0.{code}" else: secid = f"1.{code}" end_date = datetime.now().strftime('%Y%m%d') start_date = (datetime.now() - timedelta(days=days)).strftime('%Y%m%d') url = 'https://push2his.eastmoney.com/api/qt/stock/kline/get' params = { 'secid': secid, 'fields1': 'f1,f2,f3,f4,f5,f6', 'fields2': 'f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61', 'klt': '101', 'fqt': '1', 'beg': start_date, 'end': end_date, } for attempt in range(max_retries): try: resp = session.get(url, params=params, timeout=10) data = resp.json() klines = data.get('data', {}).get('klines', []) if not klines or len(klines) < 60: if attempt < max_retries - 1: time.sleep(0.5) continue return None records = [] for k in klines: parts = k.split(',') records.append({ 'date': parts[0], 'open': float(parts[1]), 'close': float(parts[2]), 'high': float(parts[3]), 'low': float(parts[4]), 'volume': float(parts[5]), 'amount': float(parts[6]), }) df = pd.DataFrame(records).reset_index(drop=True) return df except: if attempt < max_retries - 1: time.sleep(0.5) continue return None def get_index_data(secid="1.000001", max_retries=3): """获取指数数据用于判断市场趋势""" end_date = datetime.now().strftime('%Y%m%d') start_date = (datetime.now() - timedelta(days=365)).strftime('%Y%m%d') url = 'https://push2his.eastmoney.com/api/qt/stock/kline/get' params = { 'secid': secid, 'fields1': 'f1,f2,f3,f4,f5,f6', 'fields2': 'f51,f52,f53,f54,f55,f56,f57', 'klt': '101', 'fqt': '1', 'beg': start_date, 'end': end_date, } for attempt in range(max_retries): try: resp = session.get(url, params=params, timeout=15) data = resp.json() klines = data.get('data', {}).get('klines', []) if not klines: if attempt < max_retries - 1: time.sleep(1) continue return None records = [] for k in klines: parts = k.split(',') records.append({ 'date': parts[0], 'close': float(parts[2]), 'high': float(parts[3]), 'low': float(parts[4]), }) return pd.DataFrame(records).reset_index(drop=True) except: if attempt < max_retries - 1: time.sleep(1) continue return None def detect_all_historical_signals(df): """在所有历史数据点上检测指标，返回每个触发日期的指标列表""" if df is None or len(df) < 60: return {} close = df['close'].values high = df['high'].values low = df['low'].values open_ = df['open'].values volume = df['volume'].values dates = df['date'].values n = len(df) close_s = pd.Series(close) ma5 = calc_ma(close_s, 5).values ma10 = calc_ma(close_s, 10).values ma20 = calc_ma(close_s, 20).values ma60 = calc_ma(close_s, 60).values ma120 = calc_ma(close_s, 120).values vol_s = pd.Series(volume) vol_ma5 = calc_ma(vol_s, 5).values vol_ma10 = calc_ma(vol_s, 10).values vol_ma20 = calc_ma(vol_s, 20).values dif, dea, macd_bar = calc_macd(close_s) dif = dif.values dea = dea.values # 每个日期触发的指标 daily_signals = defaultdict(list) for i in range(60, n): if np.isnan(ma5[i]) or np.isnan(ma10[i]) or np.isnan(ma20[i]): continue sig_type = None # 价托 if i >= 1 and not np.isnan(ma5[i-1]) and not np.isnan(ma10[i-1]) and not np.isnan(ma20[i-1]): if ma5[i] > ma10[i] and ma5[i] > ma20[i] and ma10[i] > ma20[i]: for j in range(max(i-20, 1), i): if (ma5[j] > ma10[j] and ma5[j-1] <= ma10[j-1]) or \ (ma5[j] > ma20[j] and ma5[j-1] <= ma20[j-1]) or \ (ma10[j] > ma20[j] and ma10[j-1] <= ma20[j-1]): daily_signals[str(dates[i])].append({"indicator": "价托", "type": "看涨"}) break # 量托 if i >= 1 and not np.isnan(vol_ma5[i]) and not np.isnan(vol_ma10[i]) and not np.isnan(vol_ma20[i]): if vol_ma5[i] > vol_ma10[i] and vol_ma5[i] > vol_ma20[i] and vol_ma10[i] > vol_ma20[i]: for j in range(max(i-20, 1), i): if (vol_ma5[j] > vol_ma10[j] and vol_ma5[j-1] <= vol_ma10[j-1]) or \ (vol_ma10[j] > vol_ma20[j] and vol_ma10[j-1] <= vol_ma20[j-1]): daily_signals[str(dates[i])].append({"indicator": "量托", "type": "看涨"}) break # 多方炮 if i >= 2: if (close[i-2] > open_[i-2] and close[i-1] < open_[i-1] and close[i] > open_[i] and close[i] > open_[i-1]): daily_signals[str(dates[i])].append({"indicator": "多方炮", "type": "看涨"}) # 价压 if i >= 1 and not np.isnan(ma5[i-1]) and not np.isnan(ma10[i-1]) and not np.isnan(ma20[i-1]): if ma5[i] < ma10[i] and ma5[i] < ma20[i] and ma10[i] < ma20[i]: for j in range(max(i-20, 1), i): if (ma5[j] < ma10[j] and ma5[j-1] >= ma10[j-1]) or \ (ma10[j] < ma20[j] and ma10[j-1] >= ma20[j-1]): daily_signals[str(dates[i])].append({"indicator": "价压", "type": "看跌"}) break # 量压 if i >= 1 and not np.isnan(vol_ma5[i]) and not np.isnan(vol_ma10[i]) and not np.isnan(vol_ma20[i]): if vol_ma5[i] < vol_ma10[i] and vol_ma5[i] < vol_ma20[i] and vol_ma10[i] < vol_ma20[i]: for j in range(max(i-20, 1), i): if (vol_ma5[j] < vol_ma10[j] and vol_ma5[j-1] >= vol_ma10[j-1]) or \ (vol_ma10[j] < vol_ma20[j] and vol_ma10[j-1] >= vol_ma20[j-1]): daily_signals[str(dates[i])].append({"indicator": "量压", "type": "看跌"}) break # 金蜘蛛 if not np.isnan(ma5[i]) and not np.isnan(ma10[i]) and not np.isnan(ma20[i]): spread = max(abs(ma5[i]-ma10[i]), abs(ma5[i]-ma20[i]), abs(ma10[i]-ma20[i])) if spread < close[i] * 0.015 and ma5[i] > ma5[max(i-1,0)] and ma10[i] > ma10[max(i-1,0)]: daily_signals[str(dates[i])].append({"indicator": "金蜘蛛", "type": "看涨"}) # 死蜘蛛 if not np.isnan(ma5[i]) and not np.isnan(ma10[i]) and not np.isnan(ma20[i]): spread = max(abs(ma5[i]-ma10[i]), abs(ma5[i]-ma20[i]), abs(ma10[i]-ma20[i])) if spread < close[i] * 0.015 and ma5[i] < ma5[max(i-1,0)] and ma10[i] < ma10[max(i-1,0)]: daily_signals[str(dates[i])].append({"indicator": "死蜘蛛", "type": "看跌"}) # 出水芙蓉 if not np.isnan(ma60[i]) and not np.isnan(vol_ma20[i]): if close[i] > ma60[i] and open_[i] < ma60[i] and volume[i] > vol_ma20[i] * 1.5: below_before = any(close[j] < ma60[j] for j in range(max(i-10, 0), i) if not np.isnan(ma60[j])) if below_before: daily_signals[str(dates[i])].append({"indicator": "出水芙蓉", "type": "看涨"}) # 断头铡刀 if i >= 1 and not np.isnan(ma5[i]) and not np.isnan(ma10[i]) and not np.isnan(ma20[i]): if (close[i] < ma5[i] and close[i] < ma10[i] and close[i] < ma20[i] and open_[i] > ma5[i] and close[i] < open_[i]): daily_signals[str(dates[i])].append({"indicator": "断头铡刀", "type": "看跌"}) # 空方炮 if i >= 2: if (close[i-2] < open_[i-2] and close[i-1] > open_[i-1] and close[i] < open_[i] and close[i] < open_[i-1]): daily_signals[str(dates[i])].append({"indicator": "空方炮", "type": "看跌"}) # 三金叉见底 if i >= 1 and not np.isnan(dif[i]) and not np.isnan(dea[i]): window = 5 ma_cross = any(ma5[j] > ma10[j] and ma5[j-1] <= ma10[j-1] for j in range(max(i-window,1), i+1) if not np.isnan(ma5[j-1]) and not np.isnan(ma10[j-1])) vol_cross = any(vol_ma5[j] > vol_ma10[j] and vol_ma5[j-1] <= vol_ma10[j-1] for j in range(max(i-window,1), i+1) if not np.isnan(vol_ma5[j-1]) and not np.isnan(vol_ma10[j-1])) macd_cross = any(dif[j] > dea[j] and dif[j-1] <= dea[j-1] for j in range(max(i-window,1), i+1) if not np.isnan(dif[j-1]) and not np.isnan(dea[j-1])) if ma_cross and vol_cross and macd_cross: daily_signals[str(dates[i])].append({"indicator": "三金叉见底", "type": "看涨"}) # 三死叉见顶 if i >= 1 and not np.isnan(dif[i]) and not np.isnan(dea[i]): window = 5 ma_death = any(ma5[j] < ma10[j] and ma5[j-1] >= ma10[j-1] for j in range(max(i-window,1), i+1) if not np.isnan(ma5[j-1]) and not np.isnan(ma10[j-1])) vol_death = any(vol_ma5[j] < vol_ma10[j] and vol_ma5[j-1] >= vol_ma10[j-1] for j in range(max(i-window,1), i+1) if not np.isnan(vol_ma5[j-1]) and not np.isnan(vol_ma10[j-1])) macd_death = any(dif[j] < dea[j] and dif[j-1] >= dea[j-1] for j in range(max(i-window,1), i+1) if not np.isnan(dif[j-1]) and not np.isnan(dea[j-1])) if ma_death and vol_death and macd_death: daily_signals[str(dates[i])].append({"indicator": "三死叉见顶", "type": "看跌"}) # DIF上穿零线 if i >= 1 and not np.isnan(dif[i]) and not np.isnan(dif[i-1]): if dif[i] > 0 and dif[i-1] <= 0: daily_signals[str(dates[i])].append({"indicator": "DIF上穿零线", "type": "看涨"}) # DIF下穿零线 if i >= 1 and not np.isnan(dif[i]) and not np.isnan(dif[i-1]): if dif[i] < 0 and dif[i-1] >= 0: daily_signals[str(dates[i])].append({"indicator": "DIF下穿零线", "type": "看跌"}) # 水上金叉 if i >= 1 and not np.isnan(dif[i]) and not np.isnan(dea[i]): if dif[i] > dea[i] and dif[i-1] <= dea[i-1] and dif[i] > 0: daily_signals[str(dates[i])].append({"indicator": "水上金叉", "type": "看涨"}) # 水下死叉 if i >= 1 and not np.isnan(dif[i]) and not np.isnan(dea[i]): if dif[i] < dea[i] and dif[i-1] >= dea[i-1] and dif[i] < 0: daily_signals[str(dates[i])].append({"indicator": "水下死叉", "type": "看跌"}) # 均线多头排列 if not np.isnan(ma5[i]) and not np.isnan(ma10[i]) and not np.isnan(ma20[i]) and not np.isnan(ma60[i]): if ma5[i] > ma10[i] > ma20[i] > ma60[i] and close[i] > ma5[i]: if i >= 1 and not np.isnan(ma5[i-1]): if not (ma5[i-1] > ma10[i-1] > ma20[i-1] > ma60[i-1]): daily_signals[str(dates[i])].append({"indicator": "均线多头排列", "type": "看涨"}) # 均线空头排列 if not np.isnan(ma5[i]) and not np.isnan(ma10[i]) and not np.isnan(ma20[i]) and not np.isnan(ma60[i]): if ma5[i] < ma10[i] < ma20[i] < ma60[i] and close[i] < ma5[i]: if i >= 1 and not np.isnan(ma5[i-1]): if not (ma5[i-1] < ma10[i-1] < ma20[i-1] < ma60[i-1]): daily_signals[str(dates[i])].append({"indicator": "均线空头排列", "type": "看跌"}) # 老鸭头 if not np.isnan(ma60[i]) and i >= 5: neck = ma5[i] > ma60[i] and ma10[i] > ma60[i] duck_nose = False for j in range(max(i-10, 1), i): if j+1 < n and ma5[j] < ma10[j] and ma5[j+1] > ma10[j+1]: duck_nose = True break if neck and duck_nose and ma5[i] > ma10[i]: daily_signals[str(dates[i])].append({"indicator": "老鸭头", "type": "看涨"}) # 巨量长阳 if not np.isnan(vol_ma20[i]) and close[i] > open_[i]: body = (close[i] - open_[i]) / open_[i] if body > 0.05 and volume[i] > vol_ma20[i] * 2.0: daily_signals[str(dates[i])].append({"indicator": "巨量长阳", "type": "看涨"}) # 放量过头 if i >= 20 and not np.isnan(vol_ma20[i]): prev_high = max(high[max(i-20,0):i]) if close[i] > prev_high and volume[i] > vol_ma20[i] * 1.5: daily_signals[str(dates[i])].append({"indicator": "放量过头", "type": "看涨"}) return daily_signals def backtest_stock(code, name, current_signals): """对单只股票进行增强回测 - 量化多因子模型""" df = get_stock_data(code, days=730) # 2年数据 if df is None or len(df) < 60: return None # 在所有历史数据上检测指标 daily_signals = detect_all_historical_signals(df) if not daily_signals: return None close = df['close'].values high = df['high'].values low = df['low'].values volume = df['volume'].values dates = df['date'].values n = len(df) date_to_idx = {str(d): i for i, d in enumerate(dates)} # ===== 计算量化因子 ===== close_s = pd.Series(close) high_s = pd.Series(high) low_s = pd.Series(low) vol_s = pd.Series(volume) rsi = calc_rsi(close_s).values bb_upper, bb_lower, bb_width, bb_pctb = calc_bollinger(close_s) bb_upper = bb_upper.values bb_lower = bb_lower.values bb_pctb = bb_pctb.fillna(0.5).values atr = calc_atr(high_s, low_s, close_s).values adx, plus_di, minus_di = calc_adx(high_s, low_s, close_s) adx = adx.values vol_ratio = calc_volume_ratio(vol_s).fillna(1.0).values # ===== 对每个指标统计多时间窗口表现 ===== indicator_stats = defaultdict(lambda: { "total": 0, "wins": 0, "losses": 0, "returns": [], "ret_3d": [], "ret_5d": [], "mfe_5d": [], "mae_5d": [], "max_gain": 0, "max_loss": 0, "win_rate": 0, "avg_return": 0, "win_rate_3d": 0, "win_rate_5d": 0, "avg_return_3d": 0, "avg_return_5d": 0, "profit_factor": 0, "expectancy": 0, # 量化过滤后的条件胜率 "wr_rsi_filter": 0, "wr_adx_filter": 0, "wr_vol_filter": 0, "wr_all_filter": 0, "count_rsi_filter": 0, "count_adx_filter": 0, "count_vol_filter": 0, "count_all_filter": 0, }) for date_str, sigs in daily_signals.items(): idx = date_to_idx.get(date_str) if idx is None or idx + 1 >= n: continue # 多时间窗口收益 ret_1d = (close[idx + 1] / close[idx] - 1) * 100 ret_3d = (close[min(idx + 3, n - 1)] / close[idx] - 1) * 100 if idx + 3 < n else ret_1d ret_5d = (close[min(idx + 5, n - 1)] / close[idx] - 1) * 100 if idx + 5 < n else ret_3d # 5日内最大有利/不利变动 (MFE/MAE) future_highs = high[idx + 1:min(idx + 6, n)] future_lows = low[idx + 1:min(idx + 6, n)] mfe_5d = ((future_highs.max() / close[idx] - 1) * 100) if len(future_highs) > 0 else 0 mae_5d = ((future_lows.min() / close[idx] - 1) * 100) if len(future_lows) > 0 else 0 # 当时的量化因子状态 rsi_val = rsi[idx] if not np.isnan(rsi[idx]) else 50 adx_val = adx[idx] if not np.isnan(adx[idx]) else 0 vol_r = vol_ratio[idx] if not np.isnan(vol_ratio[idx]) else 1.0 bb_pos = bb_pctb[idx] if not np.isnan(bb_pctb[idx]) else 0.5 # 量化过滤条件 rsi_favorable = 30 <= rsi_val <= 65 # RSI在合理区间（非超买超卖） adx_trending = adx_val > 20 # 有趋势 vol_confirmed = vol_r > 1.2 # 放量确认 all_favorable = rsi_favorable and adx_trending and vol_confirmed for sig in sigs: ind = sig["indicator"] s = indicator_stats[ind] s["total"] += 1 s["returns"].append(ret_1d) s["ret_3d"].append(ret_3d) s["ret_5d"].append(ret_5d) s["mfe_5d"].append(mfe_5d) s["mae_5d"].append(mae_5d) if ret_1d > 0: s["wins"] += 1 else: s["losses"] += 1 s["max_gain"] = max(s["max_gain"], ret_1d) s["max_loss"] = min(s["max_loss"], ret_1d) # 量化过滤后的条件胜率 if rsi_favorable: s["count_rsi_filter"] += 1 if ret_1d > 0: s["wr_rsi_filter"] += 1 if adx_trending: s["count_adx_filter"] += 1 if ret_1d > 0: s["wr_adx_filter"] += 1 if vol_confirmed: s["count_vol_filter"] += 1 if ret_1d > 0: s["wr_vol_filter"] += 1 if all_favorable: s["count_all_filter"] += 1 if ret_1d > 0: s["wr_all_filter"] += 1 # ===== 计算统计指标 ===== for ind, s in indicator_stats.items(): total = max(s["total"], 1) s["win_rate"] = round(s["wins"] / total * 100, 1) s["avg_return"] = round(sum(s["returns"]) / total, 2) s["win_rate_3d"] = round(sum(1 for r in s["ret_3d"] if r > 0) / total * 100, 1) s["win_rate_5d"] = round(sum(1 for r in s["ret_5d"] if r > 0) / total * 100, 1) s["avg_return_3d"] = round(sum(s["ret_3d"]) / total, 2) s["avg_return_5d"] = round(sum(s["ret_5d"]) / total, 2) # 盈亏比和期望值 gains = [r for r in s["returns"] if r > 0] losses = [abs(r) for r in s["returns"] if r <= 0] gross_profit = sum(gains) if gains else 0 gross_loss = sum(losses) if losses else 1 if gross_loss == 0: gross_loss = 0.01 # 防止除以零导致 inf s["profit_factor"] = round(min(gross_profit / gross_loss, 99.99), 2) # 限制最大值防止 inf avg_win = sum(gains) / len(gains) if gains else 0 avg_loss = sum(losses) / len(losses) if losses else 0 s["expectancy"] = round((s["wins"] / total * avg_win - s["losses"] / total * avg_loss), 2) # 5日MFE/MAE s["avg_mfe_5d"] = round(sum(s["mfe_5d"]) / total, 2) s["avg_mae_5d"] = round(sum(s["mae_5d"]) / total, 2) s["risk_reward_5d"] = round(s["avg_mfe_5d"] / max(abs(s["avg_mae_5d"]), 0.01), 2) # 量化过滤后的条件胜率 s["wr_rsi_filter"] = round(s["wr_rsi_filter"] / max(s["count_rsi_filter"], 1) * 100, 1) s["wr_adx_filter"] = round(s["wr_adx_filter"] / max(s["count_adx_filter"], 1) * 100, 1) s["wr_vol_filter"] = round(s["wr_vol_filter"] / max(s["count_vol_filter"], 1) * 100, 1) s["wr_all_filter"] = round(s["wr_all_filter"] / max(s["count_all_filter"], 1) * 100, 1) # 清理临时数据 del s["returns"] del s["ret_3d"] del s["ret_5d"] del s["mfe_5d"] del s["mae_5d"] # ===== 综合回测结果：当前触发的所有指标的加权平均 ===== current_indicator_names = set(s["indicator"] for s in current_signals) total_weight = 0 weighted_win_rate = 0 weighted_return = 0 weighted_wr_3d = 0 weighted_wr_5d = 0 weighted_wr_filtered = 0 weighted_pf = 0 weighted_exp = 0 weighted_mfe = 0 weighted_mae = 0 backtested_count = 0 filtered_count = 0 for ind_name in current_indicator_names: if ind_name in indicator_stats: s = indicator_stats[ind_name] weight = s["total"] weighted_win_rate += s["win_rate"] * weight weighted_return += s["avg_return"] * weight weighted_wr_3d += s["win_rate_3d"] * weight weighted_wr_5d += s["win_rate_5d"] * weight # 使用量化过滤后的胜率（取最优过滤条件的胜率） best_filtered_wr = max(s["wr_rsi_filter"], s["wr_adx_filter"], s["wr_vol_filter"], s["wr_all_filter"]) weighted_wr_filtered += best_filtered_wr * weight weighted_pf += s["profit_factor"] * weight weighted_exp += s["expectancy"] * weight weighted_mfe += s["avg_mfe_5d"] * weight weighted_mae += s["avg_mae_5d"] * weight total_weight += weight backtested_count += 1 if s["count_all_filter"] > 0: filtered_count += 1 w = max(total_weight, 1) overall_win_rate = round(weighted_win_rate / w, 1) overall_avg_return = round(weighted_return / w, 2) overall_wr_3d = round(weighted_wr_3d / w, 1) overall_wr_5d = round(weighted_wr_5d / w, 1) overall_wr_filtered = round(weighted_wr_filtered / w, 1) overall_pf = round(weighted_pf / w, 2) overall_exp = round(weighted_exp / w, 2) overall_mfe = round(weighted_mfe / w, 2) overall_mae = round(weighted_mae / w, 2) risk_reward = round(overall_mfe / max(abs(overall_mae), 0.01), 2) # ===== 当前量化因子状态 ===== last_idx = n - 1 current_rsi = round(float(rsi[last_idx]) if not np.isnan(rsi[last_idx]) else 50, 1) current_adx = round(float(adx[last_idx]) if not np.isnan(adx[last_idx]) else 0, 1) current_vol_ratio = round(float(vol_ratio[last_idx]) if not np.isnan(vol_ratio[last_idx]) else 1.0, 2) current_bb_pos = round(float(bb_pctb[last_idx]) if not np.isnan(bb_pctb[last_idx]) else 0.5, 2) current_atr = round(float(atr[last_idx]) / close[last_idx] * 100, 2) if not np.isnan(atr[last_idx]) else 0 # 量化因子评分 (0-100) quant_score = 50 if 30 <= current_rsi <= 65: quant_score += 10 elif current_rsi > 70 or current_rsi < 25: quant_score -= 10 if current_adx > 25: quant_score += 12 elif current_adx > 20: quant_score += 6 if current_vol_ratio > 1.5: quant_score += 8 elif current_vol_ratio > 1.2: quant_score += 4 if 0.3 <= current_bb_pos <= 0.8: quant_score += 6 elif current_bb_pos > 0.9 or current_bb_pos < 0.1: quant_score -= 6 quant_score = max(0, min(100, quant_score)) return { "code": code, "name": name, "indicator_stats": dict(indicator_stats), "overall_win_rate": overall_win_rate, "overall_avg_return": overall_avg_return, "overall_wr_3d": overall_wr_3d, "overall_wr_5d": overall_wr_5d, "overall_wr_filtered": overall_wr_filtered, "overall_pf": overall_pf, "overall_exp": overall_exp, "overall_mfe": overall_mfe, "overall_mae": overall_mae, "risk_reward": risk_reward, "backtested_indicators": backtested_count, "total_historical_triggers": sum(s["total"] for s in indicator_stats.values()), # 当前量化因子状态 "current_rsi": current_rsi, "current_adx": current_adx, "current_vol_ratio": current_vol_ratio, "current_bb_pos": current_bb_pos, "current_atr_pct": current_atr, "quant_score": quant_score, } INDEX_CONFIG = [ ("1.000001", "上证指数"), ("0.399001", "深证成指"), ("0.399006", "创业板指"), ("1.000688", "科创50"), ("0.899050", "北证50"), ] def analyze_market_trend(): """分析市场趋势，判断是否单边下行（含多指数概览）""" all_indices = [] main_result = {"trend": "未知", "is_downtrend": False, "is_severe_downtrend": False, "description": "无法获取指数数据"} for secid, name in INDEX_CONFIG: df = get_index_data(secid) if df is None or len(df) < 60: all_indices.append({"name": name, "code": secid, "trend": "数据不足"}) continue close = df['close'] ma5 = calc_ma(close, 5).iloc[-1] ma10 = calc_ma(close, 10).iloc[-1] ma20 = calc_ma(close, 20).iloc[-1] ma60 = calc_ma(close, 60).iloc[-1] current = close.iloc[-1] prev = close.iloc[-2] if len(close) >= 21: ret_20d = (current / close.iloc[-21] - 1) * 100 else: ret_20d = 0 change = (current / prev - 1) * 100 if current < ma20 and current < ma60 and ma5 < ma10 < ma20: idx_trend = "下行" if ret_20d < -5: idx_trend = "单边下行" elif current > ma20 and current > ma60 and ma5 > ma10 > ma20: idx_trend = "上行" else: idx_trend = "震荡" all_indices.append({ "name": name, "code": secid, "index_close": round(current, 2), "index_change": round(change, 2), "ma5": round(ma5, 2), "ma10": round(ma10, 2), "ma20": round(ma20, 2), "ma60": round(ma60, 2), "ret_20d": round(ret_20d, 2), "trend": idx_trend, }) if secid == "1.000001": is_severe_downtrend = False is_downtrend = False trend = "震荡" description = "" if current < ma20 and current < ma60 and ma5 < ma10 < ma20: is_downtrend = True trend = "下行" if ret_20d < -5: is_severe_downtrend = True trend = "单边下行" description = f"市场处于单边下行趋势，20日跌幅{ret_20d:.1f}%，建议谨慎操作" else: description = f"市场处于下行趋势，20日跌幅{ret_20d:.1f}%" elif current > ma20 and current > ma60 and ma5 > ma10 > ma20: trend = "上行" description = f"市场处于上行趋势，20日涨幅{ret_20d:.1f}%" else: description = f"市场处于震荡格局，20日涨跌幅{ret_20d:.1f}%" main_result = { "trend": trend, "is_downtrend": is_downtrend, "is_severe_downtrend": is_severe_downtrend, "description": description, "index_close": round(current, 2), "index_change": round(change, 2), "ma5": round(ma5, 2), "ma20": round(ma20, 2), "ma60": round(ma60, 2), "ret_20d": round(ret_20d, 2), } main_result["indices"] = all_indices return main_result def generate_trading_signal(stock, backtest, market): """根据回测结果、量化因子和市场趋势生成交易信号 增强策略逻辑（量化多因子模型）： - 使用量化过滤后的条件胜率(wr_filtered)替代原始胜率，提升信号质量 - 结合盈亏比(profit_factor)和期望收益(expectancy)判断策略可行性 - 量化因子评分(quant_score)作为额外过滤条件 - ATR动态止损替代固定百分比止损 - 3日/5日胜率辅助判断持仓周期 """ # 基础回测指标 win_rate = backtest["overall_win_rate"] if backtest else 0 avg_return = backtest["overall_avg_return"] if backtest else 0 # 量化增强指标 wr_filtered = backtest.get("overall_wr_filtered", win_rate) if backtest else 0 wr_3d = backtest.get("overall_wr_3d", 0) if backtest else 0 wr_5d = backtest.get("overall_wr_5d", 0) if backtest else 0 profit_factor = backtest.get("overall_pf", 0) if backtest else 0 expectancy = backtest.get("overall_exp", 0) if backtest else 0 risk_reward = backtest.get("risk_reward", 0) if backtest else 0 quant_score = backtest.get("quant_score", 50) if backtest else 50 current_rsi = backtest.get("current_rsi", 50) if backtest else 50 current_adx = backtest.get("current_adx", 0) if backtest else 0 current_vol_ratio = backtest.get("current_vol_ratio", 1.0) if backtest else 1.0 current_atr_pct = backtest.get("current_atr_pct", 2.0) if backtest else 2.0 sentiment = stock.get("sentiment_score", 0) confidence = stock.get("confidence", 50) is_down = market.get("is_downtrend", False) is_severe = market.get("is_severe_downtrend", False) current_price = stock.get("latest_price", 0) has_backtest = backtest is not None # 使用量化过滤后的胜率作为主要决策依据 effective_wr = wr_filtered if has_backtest and wr_filtered > 0 else win_rate # 量化因子加成：quant_score >= 60 时胜率+3, < 40 时胜率-3 wr_adjusted = effective_wr + (3 if quant_score >= 60 else -3 if quant_score < 40 else 0) signal = "观望" reason = "" entry_price = 0 stop_loss = 0 target_price = 0 position_size = 0 holding_period = "1-3天" # === 买入条件 === if sentiment >= 0.2: if is_severe: # 单边下行市场：极高门槛 if has_backtest and wr_adjusted >= 55 and expectancy > 0.1 and profit_factor > 1.2: signal = "轻仓买入" reason = f"单边下行市场中，量化过滤胜率{wr_adjusted}%（原始{win_rate}%），盈亏比{min(profit_factor, 99.99):.2f}，期望{expectancy:+.2f}%，可轻仓博弈反弹" position_size = 20 holding_period = "1-2天" elif not has_backtest and sentiment >= 0.5: signal = "轻仓关注" reason = f"单边下行市场，无回测数据但看涨情绪较强({sentiment})，可轻仓关注" position_size = 15 else: signal = "观望" reason = f"单边下行市场，量化胜率{wr_adjusted}%不足以覆盖风险，建议观望" elif is_down: # 下行市场 if has_backtest and wr_adjusted >= 52 and expectancy > 0 and profit_factor > 1.1: signal = "小仓买入" reason = f"下行市场但量化胜率{wr_adjusted}%（原始{win_rate}%），盈亏比{min(profit_factor, 99.99):.2f}，期望{expectancy:+.2f}%，可小仓位介入" position_size = 30 holding_period = "2-3天" elif not has_backtest and sentiment >= 0.4: signal = "轻仓关注" reason = f"下行市场，无回测数据但看涨情绪较强({sentiment})，可轻仓关注" position_size = 20 else: signal = "观望" reason = f"下行市场，量化胜率{wr_adjusted}%一般，建议等待趋势明朗" else: # 正常/震荡市场 if has_backtest: # 综合评分：量化胜率 + 盈亏比 + 期望值 + 量化因子 if wr_adjusted >= 55 and sentiment >= 0.4 and expectancy > 0.2 and profit_factor > 1.3 and quant_score >= 55: signal = "买入" reason = f"量化综合评分优秀：过滤胜率{wr_adjusted}%（原始{win_rate}%），盈亏比{min(profit_factor, 99.99):.2f}，期望{expectancy:+.2f}%，量化因子{quant_score}分，建议买入" position_size = 50 holding_period = "3-5天" if wr_5d >= 55 else "1-3天" elif wr_adjusted >= 50 and sentiment >= 0.3 and expectancy > 0 and profit_factor > 1.1: signal = "适度买入" reason = f"量化胜率{wr_adjusted}%（原始{win_rate}%），盈亏比{min(profit_factor, 99.99):.2f}，期望{expectancy:+.2f}%，可适度建仓" position_size = 40 holding_period = "2-4天" elif wr_adjusted >= 47 and sentiment >= 0.2: signal = "轻仓关注" reason = f"量化胜率{wr_adjusted}%（原始{win_rate}%），盈亏比{min(profit_factor, 99.99):.2f}，可轻仓关注" position_size = 25 holding_period = "1-3天" elif win_rate >= 47 and sentiment >= 0.2: signal = "轻仓关注" reason = f"原始胜率{win_rate}%但量化因子未达标(评分{quant_score})，可轻仓关注" position_size = 20 else: if sentiment >= 0.5: signal = "轻仓关注" reason = f"无回测数据，但看涨情绪较强({sentiment})，可轻仓关注" position_size = 20 elif sentiment >= 0.3: signal = "观望" reason = f"无回测数据，看涨情绪{sentiment}，建议观望待回测确认" # === 卖出/风险条件 === elif sentiment <= -0.2: if is_severe and sentiment <= -0.3: signal = "建议清仓" reason = f"单边下行+看跌信号(情绪{sentiment})，胜率仅{win_rate}%，建议清仓规避" position_size = 0 elif is_down and sentiment <= -0.3: signal = "建议减仓" reason = f"下行市场+看跌信号(情绪{sentiment})，胜率{win_rate}%，建议减仓" position_size = 0 elif sentiment <= -0.4 or (has_backtest and win_rate <= 42) or (has_backtest and profit_factor < 0.8): signal = "注意风险" reason = f"看跌信号(情绪{sentiment})，胜率{win_rate}%，盈亏比{min(profit_factor, 99.99):.2f}，持仓者注意风险" position_size = 0 elif sentiment <= -0.2: signal = "观望" reason = f"看跌信号偏多(情绪{sentiment})，建议观望" # === ATR动态止盈止损 === if signal in ("买入", "适度买入", "小仓买入", "轻仓买入", "轻仓关注") and current_price > 0: entry_price = round(current_price, 2) # 使用ATR计算止损止盈，ATR能更好地反映个股波动特性 atr_val = current_atr_pct / 100 # 转为小数 if is_severe: stop_mult = 1.2 # 单边下行：紧止损 target_mult = 2.0 elif is_down: stop_mult = 1.5 target_mult = 2.5 elif has_backtest and wr_adjusted >= 52 and profit_factor > 1.3: stop_mult = 2.0 # 高胜率可承受更多波动 target_mult = 3.5 else: stop_mult = 1.5 # 保守 target_mult = 2.5 stop_loss = round(current_price * (1 - atr_val * stop_mult), 2) target_price = round(current_price * (1 + atr_val * target_mult), 2) # === 默认观望原因 === if signal == "观望" and not reason: if is_severe: reason = f"单边下行市场，建议整体观望，等待市场企稳" elif is_down: reason = f"下行市场，建议谨慎观望" elif has_backtest: reason = f"量化胜率{wr_adjusted}%（原始{win_rate}%），情绪{sentiment}，盈亏比{min(profit_factor, 99.99):.2f}，未达买入阈值" else: reason = f"无回测数据，情绪{sentiment}，建议观望" return { "signal": signal, "reason": reason, "entry_price": entry_price, "stop_loss": stop_loss, "target_price": target_price, "position_size": position_size, "win_rate": win_rate, "wr_filtered": wr_filtered, "wr_adjusted": wr_adjusted, "avg_return": avg_return, "profit_factor": profit_factor, "expectancy": expectancy, "quant_score": quant_score, "has_backtest": has_backtest, "holding_period": holding_period, } def main(): with open(RESULT_FILE, 'r') as f: data = json.load(f) top50 = data["top50"] print(f"开始回测 {len(top50)} 只股票...") # 1. 分析市场趋势 print("\n[1] 分析市场趋势...") market = analyze_market_trend() print(f" 市场趋势: {market['trend']} - {market['description']}") # 2. 回测每只股票 print("\n[2] 回测个股历史指标表现...") backtest_results = {} completed = 0 with ThreadPoolExecutor(max_workers=8) as executor: futures = {} for stock in top50: future = executor.submit(backtest_stock, stock["code"], stock["name"], stock["signals"]) futures[future] = stock["code"] for future in as_completed(futures): code = futures[future] completed += 1 result = future.result() if result: backtest_results[code] = result if completed % 10 == 0: print(f" 进度: {completed}/{len(top50)}") print(f" 回测完成: {len(backtest_results)}/{len(top50)} 只有历史数据") # 3. 生成交易信号 print("\n[3] 生成交易信号...") signals = {} buy_signals = [] sell_signals = [] for stock in top50: code = stock["code"] bt = backtest_results.get(code) # 从signals计算情绪评分和置信度（因为scan_results_filtered.json中不包含这些字段） stock_signals = stock.get("signals", []) bullish_count = sum(1 for s in stock_signals if s["type"] == "看涨") bearish_count = sum(1 for s in stock_signals if s["type"] == "看跌") total_count = max(len(stock_signals), 1) sentiment_score = round((bullish_count - bearish_count) / total_count, 2) confidence = min(100, max(0, 50 + bullish_count * 3 - bearish_count * 2)) stock_enriched = { **stock, "sentiment_score": sentiment_score, "confidence": confidence, } sig = generate_trading_signal(stock_enriched, bt, market) signals[code] = sig if sig["signal"] in ("买入", "适度买入", "小仓买入", "轻仓买入", "轻仓关注"): buy_signals.append({"code": code, "name": stock["name"], **sig}) elif sig["signal"] in ("建议清仓", "建议减仓", "注意风险"): sell_signals.append({"code": code, "name": stock["name"], **sig}) # 按胜率排序 buy_signals.sort(key=lambda x: x["win_rate"], reverse=True) sell_signals.sort(key=lambda x: x["win_rate"]) print(f" 买入信号: {len(buy_signals)} 只") print(f" 卖出信号: {len(sell_signals)} 只") print(f" 观望: {len(top50) - len(buy_signals) - len(sell_signals)} 只") # 4. 保存结果 output = { "scan_date": data["scan_date"], "backtest_date": datetime.now().strftime('%Y-%m-%d %H:%M'), "market_trend": market, "backtest_results": backtest_results, "trading_signals": signals, "buy_signals": buy_signals, "sell_signals": sell_signals, } with open(OUTPUT_FILE, 'w', encoding='utf-8') as f: import math def _sanitize(obj): if isinstance(obj, float): if math.isinf(obj): return 99.99 if obj > 0 else 0.0 if math.isnan(obj): return 0.0 return obj elif isinstance(obj, dict): return {k: _sanitize(v) for k, v in obj.items()} elif isinstance(obj, list): return [_sanitize(i) for i in obj] return obj json.dump(_sanitize(output), f, ensure_ascii=False, indent=2) print(f"\n回测结果已保存到 {OUTPUT_FILE}") # 打印摘要 print(f"\n{'='*60}") print(f"市场趋势: {market['trend']} - {market['description']}") print(f"\n买入信号 ({len(buy_signals)} 只):") for s in buy_signals[:5]: print(f" {s['name']} | 信号:{s['signal']} | 胜率:{s['win_rate']}% | " f"入场:{s['entry_price']} | 止损:{s['stop_loss']} | 目标:{s['target_price']} | 仓位:{s['position_size']}%") if sell_signals: print(f"\n卖出信号 ({len(sell_signals)} 只):") for s in sell_signals[:5]: print(f" {s['name']} | 信号:{s['signal']} | 胜率:{s['win_rate']}% | {s['reason']}") if __name__ == "__main__": main()