我的一大堆文件就这样消失了？？？？重新登陆也不行！ -- coding: utf-8 -- """ 同花顺SuperMind量化选股策略 基本面选股 + 30%止盈 + 8%止损 + 自动调仓 + 备选宽松条件 + 股票黑名单 核心规则： 市值20亿~500亿，优质财务指标选股，剔除ST/*ST 每日开盘前选股，最多持仓50只，等权分配资金 单只个股持仓成本价对比，涨幅达到30%止盈清仓，跌幅达到8%止损清仓 严格条件选股数量不足时自动放宽财务门槛，防止长期空仓 """ import pandas as pd 全局自定义参数（单位：元） MIN_MARKET_VALUE = 20 * 108 # 最小市值20亿 MAX_MARKET_VALUE = 500 * 108 # 最大市值500亿 MAX_SELECT_NUM = 50 # 最多持仓50只股票 BLACK_LIST = [] # 自定义股票黑名单，示例：["600000.XSHG"] 选股过少时的宽松备用阈值 RELAX_ROE = 10 RELAX_GROSS_MARGIN = 25 RELAX_REVENUE_YOY = 5 止盈止损比例 PROFIT_RATE = 0.30 LOSS_RATE = -0.08 def init(context): """策略初始化函数""" context.hold_stocks = [] context.target_stocks = [] # 新增：盘前选股结果缓存 context.cost_price = dict() # 记录每只股票建仓成本价 def before_trading_start(context, bar_dict): """每个交易日开盘前执行选股，只做筛选，不交易""" def get_stock_data(roe_limit, margin_limit, yoy_limit): q = query( supermind_valuation.code, supermind_valuation.market_cap, supermind_fin_indicator.net_profit_parent, supermind_fin_indicator.operating_revenue, supermind_fin_indicator.roe, supermind_fin_indicator.gross_margin_ratio, supermind_valuation.pe_ttm, supermind_fin_indicator.revenue_yoy ).filter( supermind_valuation.market_cap >= MIN_MARKET_VALUE, supermind_valuation.market_cap <= MAX_MARKET_VALUE, supermind_fin_indicator.net_profit_parent > 0, supermind_fin_indicator.operating_revenue > 1 * 10**8, supermind_fin_indicator.roe > roe_limit, supermind_fin_indicator.gross_margin_ratio > margin_limit, supermind_fin_indicator.revenue_yoy > yoy_limit, supermind_valuation.pe_ttm >= 0, supermind_valuation.pe_ttm <= 30, ) df = get_fundamentals(q, date=context.current_dt.date()) df = df.dropna() df = df[~df["code"].isin(BLACK_LIST)] 手动过滤ST风险警示股票（标准方案） df = df[~df['code'].str.match(r'^[0,3,6]{6}[.XSHE|.XSHG]*ST')] return df 严格条件筛选 df_result = get_stock_data(15, 30, 10) if len(df_result) < 5: print("严格筛选标的不足，自动放宽财务筛选条件") df_result = get_stock_data(RELAX_ROE, RELAX_GROSS_MARGIN, RELAX_REVENUE_YOY) if df_result.empty: print("当前筛选条件下无符合要求的股票，全部空仓") target_stocks = [] else: df_result = df_result.head(MAX_SELECT_NUM) raw_codes = df_result["code"].tolist() temp_stocks = [] for code in raw_codes: if code.startswith(('0','3')): temp_stocks.append(code + '.XSHE') elif code.startswith('6'): temp_stocks.append(code + '.XSHG') target_stocks = temp_stocks print(f"筛选出{len(target_stocks)}只标的股票：{target_stocks}") 仅缓存选股结果，移到handle_bar做调仓 context.target_stocks = target_stocks context.hold_stocks = target_stocks def rebalance_position(context, bar_dict, target_codes): """等权仓位分配、自动调仓平仓函数""" total_target_num = len(target_codes) current_hold = set(context.portfolio.positions.keys()) target_set = set(target_codes) 调出池的个股清仓 for stock_code in current_hold: if stock_code not in target_set and stock_code != "cash": order_target_percent(stock_code, 0) print(f"调出选股池，卖出 {stock_code}") if stock_code in context.cost_price: del context.cost_price[stock_code] if total_target_num == 0: return single_weight = 1.0 / total_target_num 执行仓位配置，建仓时记录成本价（使用当日收盘价） for stock_code in target_codes: order_target_percent(stock_code, single_weight) print(f"调仓买入 {stock_code}，单只仓位：{single_weight:.2%}") if stock_code not in context.cost_price: # 使用当前bar收盘价作为建仓成本，解决盘前取不到价格的问题 context.cost_price[stock_code] = bar_dict[stock_code].close def handle_bar(context, bar_dict): """主线程：先执行每日调仓，再监控止盈止损""" 第一步：每日开盘后执行调仓 rebalance_position(context, bar_dict, context.target_stocks) 第二步：监控止盈止损 current_positions = context.portfolio.positions for stock_code in list(current_positions.keys()): if stock_code == "cash": continue if stock_code not in context.cost_price: continue cost = context.cost_price[stock_code] current_price = bar_dict[stock_code].close profit_ratio = (current_price - cost) / cost 30%止盈 if profit_ratio >= PROFIT_RATE: print(f"{stock_code} 收益率{profit_ratio:.2%}，触发30%止盈，清仓") order_target_percent(stock_code, 0) del context.cost_price[stock_code] if stock_code in context.hold_stocks: context.hold_stocks.remove(stock_code) 8%止损 elif profit_ratio <= LOSS_RATE: print(f"{stock_code} 收益率{profit_ratio:.2%}，触发8%止损，清仓") order_target_percent(stock_code, 0) del context.cost_price[stock_code] if stock_code in context.hold_stocks: context.hold_stocks.remove(stock_code) -- coding: utf-8 -- """ 同花顺SuperMind量化选股策略 基本面选股 + 30%止盈 + 8%止损 + 自动调仓 + 备选宽松条件 + 股票黑名单 核心规则： 市值20亿~500亿，优质财务指标选股，剔除ST/*ST 每日开盘前选股，最多持仓50只，等权分配资金 单只个股持仓成本价对比，涨幅达到30%止盈清仓，跌幅达到8%止损清仓 严格条件选股数量不足时自动放宽财务门槛，防止长期空仓 """ import pandas as pd 全局自定义参数（单位：元） MIN_MARKET_VALUE = 20 * 108 # 最小市值20亿 MAX_MARKET_VALUE = 500 * 108 # 最大市值500亿 MAX_SELECT_NUM = 50 # 最多持仓50只股票 BLACK_LIST = [] # 自定义股票黑名单，示例：["600000.XSHG"] 选股过少时的宽松备用阈值 RELAX_ROE = 10 RELAX_GROSS_MARGIN = 25 RELAX_REVENUE_YOY = 5 止盈止损比例 PROFIT_RATE = 0.30 LOSS_RATE = -0.08 def init(context): """策略初始化函数""" context.hold_stocks = [] context.target_stocks = [] # 新增：盘前选股结果缓存 context.cost_price = dict() # 记录每只股票建仓成本价 def before_trading_start(context, bar_dict): """每个交易日开盘前执行选股，只做筛选，不交易""" def get_stock_data(roe_limit, margin_limit, yoy_limit): q = query( supermind_valuation.code, supermind_valuation.market_cap, supermind_fin_indicator.net_profit_parent, supermind_fin_indicator.operating_revenue, supermind_fin_indicator.roe, supermind_fin_indicator.gross_margin_ratio, supermind_valuation.pe_ttm, supermind_fin_indicator.revenue_yoy ).filter( supermind_valuation.market_cap >= MIN_MARKET_VALUE, supermind_valuation.market_cap <= MAX_MARKET_VALUE, supermind_fin_indicator.net_profit_parent > 0, supermind_fin_indicator.operating_revenue > 1 * 10**8, supermind_fin_indicator.roe > roe_limit, supermind_fin_indicator.gross_margin_ratio > margin_limit, supermind_fin_indicator.revenue_yoy > yoy_limit, supermind_valuation.pe_ttm >= 0, supermind_valuation.pe_ttm <= 30, ) df = get_fundamentals(q, date=context.current_dt.date()) df = df.dropna() df = df[~df["code"].isin(BLACK_LIST)] 手动过滤ST风险警示股票（标准方案） df = df[~df['code'].str.match(r'^[0,3,6]{6}[.XSHE|.XSHG]*ST')] return df 严格条件筛选 df_result = get_stock_data(15, 30, 10) if len(df_result) < 5: print("严格筛选标的不足，自动放宽财务筛选条件") df_result = get_stock_data(RELAX_ROE, RELAX_GROSS_MARGIN, RELAX_REVENUE_YOY) if df_result.empty: print("当前筛选条件下无符合要求的股票，全部空仓") target_stocks = [] else: df_result = df_result.head(MAX_SELECT_NUM) raw_codes = df_result["code"].tolist() temp_stocks = [] for code in raw_codes: if code.startswith(('0','3')): temp_stocks.append(code + '.XSHE') elif code.startswith('6'): temp_stocks.append(code + '.XSHG') target_stocks = temp_stocks print(f"筛选出{len(target_stocks)}只标的股票：{target_stocks}") 仅缓存选股结果，移到handle_bar做调仓 context.target_stocks = target_stocks context.hold_stocks = target_stocks def rebalance_position(context, bar_dict, target_codes): """等权仓位分配、自动调仓平仓函数""" total_target_num = len(target_codes) current_hold = set(context.portfolio.positions.keys()) target_set = set(target_codes) 调出池的个股清仓 for stock_code in current_hold: if stock_code not in target_set and stock_code != "cash": order_target_percent(stock_code, 0) print(f"调出选股池，卖出 {stock_code}") if stock_code in context.cost_price: del context.cost_price[stock_code] if total_target_num == 0: return single_weight = 1.0 / total_target_num 执行仓位配置，建仓时记录成本价（使用当日收盘价） for stock_code in target_codes: order_target_percent(stock_code, single_weight) print(f"调仓买入 {stock_code}，单只仓位：{single_weight:.2%}") if stock_code not in context.cost_price: # 使用当前bar收盘价作为建仓成本，解决盘前取不到价格的问题 context.cost_price[stock_code] = bar_dict[stock_code].close def handle_bar(context, bar_dict): """主线程：先执行每日调仓，再监控止盈止损""" 第一步：每日开盘后执行调仓 rebalance_position(context, bar_dict, context.target_stocks) 第二步：监控止盈止损 current_positions = context.portfolio.positions for stock_code in list(current_positions.keys()): if stock_code == "cash": continue if stock_code not in context.cost_price: continue cost = context.cost_price[stock_code] current_price = bar_dict[stock_code].close profit_ratio = (current_price - cost) / cost 30%止盈 if profit_ratio >= PROFIT_RATE: print(f"{stock_code} 收益率{profit_ratio:.2%}，触发30%止盈，清仓") order_target_percent(stock_code, 0) del context.cost_price[stock_code] if stock_code in context.hold_stocks: context.hold_stocks.remove(stock_code) 8%止损 elif profit_ratio <= LOSS_RATE: print(f"{stock_code} 收益率{profit_ratio:.2%}，触发8%止损，清仓") order_target_percent(stock_code, 0) del context.cost_price[stock_code] if stock_code in context.hold_stocks: context.hold_stocks.remove(stock_code) 初学写策略代码，跑回测无成交。请老师们指教 我回测了 A 股 10 年的"追涨停"策略，结果可能和你想的不一样 A 股散户圈里有一种玩法叫"打板"——今天某只股票涨停了，明天开盘就买进去，赌它继续涨。 打板选手有一套自己的逻辑：涨停说明有资金强势介入，次日惯性上冲的概率不低，如果能接力涨停就赚 10%，不能涨停就止损出来。听起来赔率不错。 但"听起来"和"跑数据"是两回事。 这篇文章用 AlphaFeed 拉 A 股历史数据，做一个严格的统计回测：涨停之后的第二天、第三天、第五天，平均涨跌是多少？胜率多高？如果严格执行"涨停次日买入"策略，长期下来是赚是亏？ 先说结论：追涨停长期来看大概率亏钱，但细节比你以为的复杂。 1. 找出历史上所有涨停的日子 A 股涨停幅度一般是 10%（创业板/科创板 20%），但实际因为四舍五入，涨幅在 9.5%–10.5% 之间就可以算涨停。 我们先选一组活跃股，拉长期 K 线，找出所有涨停日： import pandas as pd from alphafeed import AlphaFeed af = AlphaFeed() symbols = [ "000001.SZ", "600519.SH", "000858.SZ", "601318.SH", "002594.SZ", "300750.SZ", "600036.SH", "000333.SZ", "601012.SH", "600276.SH", "000568.SZ", "601888.SH", "002415.SZ", "300059.SZ", "600809.SH", "000651.SZ", "002304.SZ", "300124.SZ", "002475.SZ", "600887.SH", ] klines = af.klines.batch( symbols, period="1d", count=5000, adjust="none", # 用不复权数据，否则涨跌幅会被复权扭曲 to_dataframe=True, show_progress=True, ) all_limit_ups = [] for sym, df in klines.items(): df = df.sort_values("trade_date").reset_index(drop=True) df["ret"] = df["close"].pct_change() for i in range(1, len(df)): if df["ret"].iloc[i] >= 0.095: # 涨停（含误差） all_limit_ups.append({ "symbol": sym, "date": df["trade_date"].iloc[i], "close": df["close"].iloc[i], "idx": i, }) print(f"共找到 {len(all_limit_ups)} 次涨停") print(f"覆盖 {len(klines)} 只标的") 2. 涨停次日表现：平均能赚多少？ 核心问题：今天涨停，明天买进去，一天能赚多少？ import pandas as pd import numpy as np from alphafeed import AlphaFeed af = AlphaFeed() symbols = [ "000001.SZ", "600519.SH", "000858.SZ", "601318.SH", "002594.SZ", "300750.SZ", "600036.SH", "000333.SZ", "601012.SH", "600276.SH", "000568.SZ", "601888.SH", "002415.SZ", "300059.SZ", "600809.SH", "000651.SZ", ] klines = af.klines.batch( symbols, period="1d", count=5000, adjust="none", to_dataframe=True, show_progress=True, ) results = [] for sym, df in klines.items(): df = df.sort_values("trade_date").reset_index(drop=True) df["ret"] = df["close"].pct_change() for i in range(1, len(df) - 5): if df["ret"].iloc[i] < 0.095: continue # 涨停次日买入（用次日开盘价） if i + 1 >= len(df): continue buy_price = df["open"].iloc[i + 1] if buy_price <= 0: continue # 次日收益（用次日收盘价） ret_1d = df["close"].iloc[i + 1] / buy_price - 1 # 第3日收益 ret_2d = df["close"].iloc[i + 2] / buy_price - 1 if i + 2 < len(df) else None # 第5日收益 ret_5d = df["close"].iloc[i + 5] / buy_price - 1 if i + 5 < len(df) else None # 是否连板（次日也涨停） next_ret = df["ret"].iloc[i + 1] is_连板 = next_ret >= 0.095 results.append({ "symbol": sym, "date": df["trade_date"].iloc[i], "ret_1d": ret_1d, "ret_2d": ret_2d, "ret_5d": ret_5d, "is_连板": is_连板, "open_pct": (buy_price / df["close"].iloc[i] - 1), # 次日开盘溢价 }) rdf = pd.DataFrame(results) print(f"=== 涨停次日表现统计（{len(rdf)} 次涨停样本）===") print() for label, col in [("持有1天", "ret_1d"), ("持有2天", "ret_2d"), ("持有5天", "ret_5d")]: valid = rdf[col].dropna() print(f" {label}:") print(f" 平均收益: {valid.mean():+.2%}") print(f" 中位数: {valid.median():+.2%}") print(f" 胜率: {(valid > 0).mean():.1%}") print(f" 最大赚: {valid.max():+.2%}") print(f" 最大亏: {valid.min():+.2%}") print() 你可能会看到的结果 在多数统计中，涨停次日的平均收益接近 0 或微负，但中位数通常为负。这意味着： 少数大赚的案例拉高了平均值（连板翻倍的那些） 多数情况是小亏：高开低走、开盘冲高后回落 胜率大概在 40%–50% 之间——连抛硬币都不如 3. 次日开盘溢价：买入成本被抬高了 涨停次日最大的问题不是"涨不涨"，而是你买不到好价格。 因为昨天涨停了，今天大家都想买，开盘价往往直接高开 3%–5%。你以为自己在追涨停，其实你在高位接盘： print("=== 涨停次日开盘溢价分布 ===") print(f" 平均开盘溢价: {rdf['open_pct'].mean():+.2%}") print(f" 中位数溢价: {rdf['open_pct'].median():+.2%}") print() bins = [(-1, -0.02), (-0.02, 0), (0, 0.02), (0.02, 0.05), (0.05, 0.10), (0.10, 1)] labels = ["低开>2%", "低开0-2%", "高开0-2%", "高开2-5%", "高开5-10%", "高开>10%"] for (lo, hi), label in zip(bins, labels): pct = ((rdf["open_pct"] >= lo) & (rdf["open_pct"] < hi)).mean() print(f" {label}: {pct:.1%}") 如果次日平均高开 3%，那你买入的一瞬间已经"亏了 3%"。即使当天收涨 2%，你的实际收益是 -1%。 这就是追涨停最大的隐性成本：你的买入价不是昨天的涨停价，而是今天被抬高后的开盘价。 4. 连板概率：买到第二个涨停有多难 打板选手追求的最大收益来源是"连板"——买进去之后第二天继续涨停。我们来看看概率： 连板率 = rdf["is_连板"].mean() print(f"=== 连板概率 ===") print(f" 涨停后次日再涨停的概率: {连板率:.1%}") print(f" 也就是说，每 {1/连板率:.0f} 次涨停里，大约有 1 次连板") print() # 连板 vs 不连板的收益差异 连板组 = rdf[rdf["is_连板"]] 非连板组 = rdf[~rdf["is_连板"]] print(f" 连板组 (n={len(连板组)}):") print(f" 次日收益: {连板组['ret_1d'].mean():+.2%}") print(f" 非连板组 (n={len(非连板组)}):") print(f" 次日收益: {非连板组['ret_1d'].mean():+.2%}") 一般来说，连板概率在 10%–20% 之间。这意味着你打 10 次板，可能只有 1–2 次吃到连板的肉，其余 8 次都在亏开盘溢价和高开低走的差价。 5. 什么样的涨停更值得追 不是所有涨停都一样。我们可以按几个维度拆分，看看哪种涨停次日表现更好： import pandas as pd import numpy as np from alphafeed import AlphaFeed af = AlphaFeed() # 假设已经有 rdf（上面的结果 DataFrame） # 维度1: 成交额大小 # 需要在构建 results 时加入成交额字段，这里补充 symbols = list(set(rdf["symbol"].tolist())) klines = af.klines.batch( symbols, period="1d", count=5000, adjust="none", to_dataframe=True, ) enriched = [] for _, row in rdf.iterrows(): sym = row["symbol"] if sym not in klines: continue kdf = klines[sym] kdf = kdf.sort_values("trade_date").reset_index(drop=True) match = kdf[kdf["trade_date"] == row["date"]] if match.empty: continue amount = match["amount"].values[0] volume = match["volume"].values[0] prev_amount = kdf[kdf["trade_date"] < row["date"]]["amount"].tail(20).mean() vol_ratio = amount / prev_amount if prev_amount > 0 else 1 enriched.append({ **row.to_dict(), "amount": amount, "vol_ratio": vol_ratio, }) edf = pd.DataFrame(enriched) print("=== 按成交额分组的涨停次日表现 ===") edf["amount_group"] = pd.qcut(edf["amount"], q=3, labels=["低成交额", "中成交额", "高成交额"]) for group in ["低成交额", "中成交额", "高成交额"]: g = edf[edf["amount_group"] == group] print(f" {group} (n={len(g)}): 次日均值{g['ret_1d'].mean():+.2%} 胜率{(g['ret_1d']>0).mean():.0%}") print() print("=== 按量比分组的涨停次日表现 ===") edf["vol_group"] = pd.cut(edf["vol_ratio"], bins=[0, 1.5, 3, 100], labels=["量比<1.5", "量比1.5-3", "量比>3"]) for group in ["量比<1.5", "量比1.5-3", "量比>3"]: g = edf[edf["vol_group"] == group] if len(g) < 5: continue print(f" {group} (n={len(g)}): 次日均值{g['ret_1d'].mean():+.2%} 胜率{(g['ret_1d']>0).mean():.0%}") 通常的发现是： 涨停类型 次日表现 解读 缩量涨停（量比低） 相对较好 筹码锁定好，抛压小 放巨量涨停（量比>3） 相对较差 换手大，可能是拉高出货 高成交额涨停 不确定 大盘股涨停少见，统计上有参考价值 低成交额涨停 波动大 小盘股弹性高但风险也高 6. 止损纪律：不止损会怎样 打板选手最重要的规则之一是"错了就砍"。不止损的后果有多严重？我们来对比： import pandas as pd import numpy as np # 模拟三种执行方式 strategies = { "无止损（持有5天）": rdf["ret_5d"].dropna(), "止损-3%": rdf["ret_1d"].apply(lambda x: x if x > -0.03 else -0.03), "止损-5%": rdf["ret_1d"].apply(lambda x: x if x > -0.05 else -0.05), } print("=== 止损 vs 不止损 ===") for name, rets in strategies.items(): equity = (1 + rets).cumprod() total_ret = equity.iloc[-1] - 1 max_dd = (equity / equity.cummax() - 1).min() print(f" {name}:") print(f" 累计收益: {total_ret:+.2%}") print(f" 最大回撤: {max_dd:.2%}") print(f" 平均单次: {rets.mean():+.2%}") print() 不止损的结果通常非常难看——因为那些大亏的 case（涨停次日直接跌 5%–8%）会把利润全部吃掉。严格止损会改善结果，但可能从"大亏"变成"小亏"。 7. 累计净值曲线：长期跑下来什么样 把每一次"涨停次日买入、收盘卖出"的收益串成净值曲线： import pandas as pd import numpy as np rdf_sorted = rdf.sort_values("date").reset_index(drop=True) equity = [1.0] for _, row in rdf_sorted.iterrows(): ret = row["ret_1d"] if pd.isna(ret): continue cost = 0.0015 # 单边手续费+印花税 net_ret = ret - cost * 2 equity.append(equity[-1] * (1 + net_ret)) equity_series = pd.Series(equity) total_return = equity_series.iloc[-1] - 1 max_dd = (equity_series / equity_series.cummax() - 1).min() print(f"=== 追涨停策略长期表现 ===") print(f" 交易次数: {len(rdf_sorted)}") print(f" 累计收益: {total_return:+.2%}") print(f" 最大回撤: {max_dd:.2%}") print(f" 最终净值: {equity_series.iloc[-1]:.4f}") print() if total_return < 0: print(f" 结论: 无差别追涨停，长期是亏钱的") else: print(f" 结论: 有正收益，但需要看夏普和回撤是否可接受") 加上交易成本之后，结果会更加难看。每次交易都要付万 3 佣金 + 千 1 印花税，打板交易频率高，成本累积很快。 8. 这意味着什么 回测结果通常指向几个结论： 1. 无差别追涨停是负期望的。 平均次日收益在扣除成本后大概率为负。涨停不是免费的"涨 10%"——你的买入成本被次日高开抬高了，而高开之后回落的概率比继续涨的概率大。 2. 少数人能赚钱，因为他们不是无差别追。 真正的打板高手会筛选：首板 vs 二板、题材强度、板块联动、分时走势、封单量。他们追的是特定条件下的涨停，而不是看到涨停就冲。 3. 追涨停赚钱的本质不是"涨停好"，而是"选股好"。 如果你能选出明天涨停的票，你当然赚钱。但这和"追涨停"是两回事。追涨停是在涨停已经发生之后买入，这时候信息已经被价格消化了。 4. 打板的收益分布是极端的。 少数连板收益很大，多数普通涨停次日是亏的。这意味着你必须交易很多次才能碰到几次大赚，但过程中的亏损会消耗你的本金和心态。 9. 如果你还是想研究打板 把上面的分析框架改成你自己的筛选条件，看看特定条件下的涨停是否表现更好： # 示例：只看"首板"（前一天不是涨停的涨停） 首板 = rdf[rdf["open_pct"] < 0.095] # 次日不是一字板，说明不是连板后的 print(f"首板次日表现 (n={len(首板)}):") print(f" 平均收益: {首板['ret_1d'].mean():+.2%}") print(f" 胜率: {(首板['ret_1d'] > 0).mean():.0%}") 你也可以加入更多筛选维度： 板块里同时涨停的个数（题材强度） 涨停时间（早盘涨停 vs 尾盘涨停） 涨停前的走势（底部涨停 vs 高位涨停） 市值区间（小盘涨停 vs 大盘涨停） 每加一个条件，样本量就会减少，但如果剩下的样本显示出稳定的正期望，那可能是一个值得深入研究的方向。 10. 你可以用这个方法验证任何"民间策略" 这篇文章的价值不只是"打板赚不赚钱"这一个结论，而是给了你一套验证方法： 提出假设 → 定义规则 → 拉历史数据 → 统计收益/胜率/回撤 → 下结论 AlphaFeed 可以拉数千根 K 线，支持不复权模式（对涨停分析很重要），批量获取几十只票的数据。有了数据，任何策略假设都可以用同样的方法验证。 下次有人跟你说"这种票明天一定涨"，你不用争论——打开终端，跑一下数据。 结语 追涨停是 A 股散户圈最有争议的话题之一。有人靠它赚到过钱（他们的故事你都听过），有人靠它亏到清仓（他们的故事你没听过）。 数据告诉我们的是：无差别追涨停的长期期望收益为负。 涨停次日高开低走的概率、交易成本的累积、连板极低的概率，这些因素加在一起，让"看到涨停就追"变成一个负期望的游戏。 但这不意味着所有涨停都不值得关注。如果你有能力在涨停中进一步筛选——通过成交量特征、板块联动、市场情绪、分时结构——那有可能找到正期望的子集。只是，这已经不是"追涨停"，而是"选股"了。 参考文献： AlphaFeed 官网：https://alphafeed.org/ Python SDK 快速开始：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart 最简单的问句就报错了。query_iwencai("市值>1000亿，日成交额>30亿，换手率大于4.5%") --------------------------------------------------------------------------- FileNotFoundError Traceback (most recent call last) /opt/conda/lib/python3.8/site-packages/mgquant_mod_mindgo/data/trading_dates_store.pyc in __init__(self, f, table, process_fn) /opt/conda/lib/python3.8/site-packages/bcolz/toplevel.py in open(rootdir, mode) 139 else: --> 140 return bcolz.carray(rootdir=rootdir, mode=mode) 141 bcolz/carray_ext.pyx in bcolz.carray_ext.carray.__cinit__() bcolz/carray_ext.pyx in bcolz.carray_ext.carray._read_meta() FileNotFoundError: [Errno 2] No such file or directory: '/home/jovyan/backtest-data/mgquant/stock/trading_dates.bcolz/meta/sizes' During handling of the above exception, another exception occurred: OSError Traceback (most recent call last) /opt/conda/lib/python3.8/site-packages/IPython/core/interactiveshell.py in transform_cell(self, raw_cell) 3207 # use prefilter_lines to handle trailing newlines 3208 # restore trailing newline for ast.parse -> 3209 cell = self.prefilter_manager.prefilter_lines(cell) + '\n' 3210 3211 lines = cell.splitlines(keepends=True) /opt/conda/lib/python3.8/site-packages/IPython/core/prefilter.py in prefilter_lines(self, lines, continue_prompt) 333 for lnum, line in enumerate(llines) ]) 334 else: --> 335 out = self.prefilter_line(llines[0], continue_prompt) 336 337 return out /opt/conda/lib/python3.8/site-packages/IPython/core/prefilter.py in prefilter_line(self, line, continue_prompt) 308 return normal_handler.handle(line_info) 309 --> 310 prefiltered = self.prefilter_line_info(line_info) 311 # print "prefiltered line: %r" % prefiltered 312 return prefiltered /opt/conda/lib/python3.8/site-packages/IPython/core/prefilter.py in prefilter_line_info(self, line_info) 250 """ 251 # print "prefilter_line_info: ", line_info --> 252 handler = self.find_handler(line_info) 253 return handler.handle(line_info) 254 /opt/conda/lib/python3.8/site-packages/IPython/core/prefilter.py in find_handler(self, line_info) 257 for checker in self.checkers: 258 if checker.enabled: --> 259 handler = checker.check(line_info) 260 if handler: 261 return handler /opt/conda/lib/python3.8/site-packages/IPython/core/prefilter.py in check(self, line_info) 414 def check(self, line_info): 415 obj = self.shell.user_ns.get(line_info.ifun) --> 416 if isinstance(obj, Macro): 417 return self.prefilter_manager.get_handler_by_name('macro') 418 else: /opt/conda/lib/python3.8/site-packages/mgquant/utils/lazyimport.pyc in __getattribute__(self, attr) /opt/conda/lib/python3.8/site-packages/mgquant/utils/lazyimport.pyc in _resolve(self) /opt/conda/lib/python3.8/site-packages/mgquant/utils/lazyimport.pyc in _import(self, scope, name) /opt/conda/lib/python3.8/site-packages/mgquant_mod_stock/research_api.pyc in <module> /opt/conda/lib/python3.8/site-packages/mgquant_mod_mindgo/research/research_api.pyc in <module> /opt/conda/lib/python3.8/site-packages/mgquant_mod_stock/data/data_proxy.pyc in __init__(self, data_source) /opt/conda/lib/python3.8/site-packages/mgquant_mod_stock/data/fast_data_source.pyc in get_trading_calendar(self) /opt/conda/lib/python3.8/site-packages/mgquant_mod_mindgo/utils/wrap_utils.pyc in __get__(self, instance, owner) /opt/conda/lib/python3.8/site-packages/mgquant_mod_stock/data/fast_data_source.pyc in _trading_dates(self) /opt/conda/lib/python3.8/site-packages/mgquant_mod_mindgo/data/trading_dates_store.pyc in __init__(self, f, table, process_fn) /opt/conda/lib/python3.8/os.py in makedirs(name, mode, exist_ok) 211 if head and tail and not path.exists(head): 212 try: --> 213 makedirs(head, exist_ok=exist_ok) 214 except FileExistsError: 215 # Defeats race condition when another thread created the path /opt/conda/lib/python3.8/os.py in makedirs(name, mode, exist_ok) 221 return 222 try: --> 223 mkdir(name, mode) 224 except OSError: 225 # Cannot rely on checking for EEXIST, since the operating system OSError: [Errno 30] Read-only file system: '/home/jovyan/backtest-data/mgquant' 请大家不要客气，任何意见建议可以在这里评论提出。 被采纳后我们将奖励1G研究环境内存 3个月。 最近好几个朋友在后台问，想研究外盘期货，数据从哪找。我一开始也头大，各种交易所格式不统一，自己爬又麻烦。后来发现一个叫CMES金融数据库的地方，东西还挺全，就把它当主要数据源了。今天不聊策略，就单纯扒拉一下它里面到底有哪些数据，每个数据文件里都装着什么字段。看完这篇，你大概就知道这些数据能不能满足你的需求了。 先说说整体感觉吧。这个数据库主要覆盖的是主流的境外期货交易所，比如大家常听的LME（伦敦金属）、CME（芝商所）、ICE（洲际）、EUREX（欧交所）这些。数据种类分得挺细，有最原始的逐笔成交（Tick），有整理好的分钟线，还有日线。对做高频、中频或者单纯想回测个长周期策略的人来说，基本都够用了。 一、 最细的粒度：逐笔成交与委托数据 如果你在研究订单流或者盘口动力学，这个数据跑不掉。但说实话，新手慎入，数据量太大了，一个活跃合约一天的数据就能轻松上G，处理起来挺费劲的。 它里面主要包含这些字段，我拿CME的ES（标普500指数期货）举个例子： timestamp (时间戳)： 精确到毫秒的成交时间。这是所有分析的基准，对齐不同数据源就靠它。 price (成交价)： 这一笔交易达成的价格。 volume (成交量)： 这一笔交易的合约数量。 bid_price / ask_price (买一价/卖一价)： 这笔成交发生时的最优买卖报价。注意不是全档位。 bid_size / ask_size (买一量/卖一量)： 对应报价上的订单数量。 trade_type (交易类型)： 这个挺重要，标识是普通成交、大宗交易还是跨期价差交易等等。不同交易所的标识符可能不同。 有的交易所数据还会有**order_id（订单号）**的变化，能让你跟踪单个订单的生命周期，但这属于更精细的Level 2甚至Level 3范畴了，不是所有品种都有。 为了验证一些盘口微观结构，我调取了CMES金融数据库中过去三年的主力合约数据进行回测，清洗和匹配这些tick数据确实是体力活。如果你只是好奇，可以用他们的接口先下个小样本看看结构，下面是个简单的Python调用示例： # 示例：调用CMES数据接口获取tick数据样例 # 注意：需要先安装他们的SDK，通常 pip install cmes-sdk 就行（具体看官方文档） import cmes_client # 初始化客户端，需要你的认证密钥 client = cmes_client.Client(api_key='your_api_key_here') # 请求ES合约的tick数据，注意参数格式要严格按照文档来 # 这里请求2023年某一天的数据作为例子 try: tick_data = client.get_tick_data( exchange='CME', symbol='ES', date='2023-10-26', data_type='trade' # 获取成交数据 ) print(f"获取到 {len(tick_data)} 条tick记录") # 看看前几条长啥样 print(tick_data.head()) except Exception as e: print(f"调用出错：{e}。请检查入参是否正确，以及账户权限和调用频率限制。") 二、 更常用的：分钟线与日线数据 对于大多数不需要tick级精度的策略回测，分钟线和日线是主力。数据规整，量也小得多。 分钟线数据通常包含以下字段： 字段名 说明 备注 datetime K线起始时间 通常是每分钟的第一秒 open 分钟内的开盘价 high 分钟内的最高价 low 分钟内的最低价 close 分钟内的收盘价 volume 分钟内的总成交量 累计值 open_interest 持仓量 不是所有交易所或周期都提供，日线更常见 日线数据字段和分钟线类似，就是时间周期变成了一天。但日线数据有时会包含调整后的收盘价，用于处理合约展期、分红等事件，这对长期回测避免断层至关重要。我一开始用原始价格回测，结果在换月的时候净值曲线跳得跟心电图似的，后来换了调整后数据才正常。 三、 都覆盖哪些交易所？ 这是大家最关心的，我把自己常用的几个交易所和其主打品种列一下，不全，但主流的基本在了： LME (伦敦金属交易所)： 铜、铝、锌、镍等基础金属。它的数据结构有点特殊，因为是环形交易，但数据库里一般会处理成连续的电子盘价格。 CME Group (芝商所集团)： 这是个巨无霸，旗下包括： CME： 股指期货（标普500、纳斯达克100）、外汇、利率产品。 CBOT： 农产品（大豆、玉米、小麦）、美国国债。 NYMEX： 能源（WTI原油、天然气）、金属（黄金、白银）。 COMEX： 高级金属（黄金、白银、铜）。 ICE (洲际交易所)： 布伦特原油、糖、咖啡、可可等软商品，以及美元指数等。 EUREX (欧洲期货交易所)： 欧元区股指期货（如德国DAX）、欧洲利率产品。 SGX (新加坡交易所)： 富时中国A50指数期货，日经225指数期货等，玩亚太市场的关注。 JPX (日本交易所集团)： 日经225、TOPIX指数期货。 HKEX (香港交易所)： 恒生指数期货、H股指数期货。 每个交易所的数据细节会有差异，比如价格单位、合约代码规则、交易时间（是否包含盘前盘后）等。下载的时候最好先看看数据说明文档。 四、 一些零散但重要的点 数据质量： 历史数据难免有异常值或缺失。这个数据库的数据是经过基础清洗的，比如过滤明显错误的价格，但使用者最好还是自己再做一遍简单的合理性检查，比如价格跳变是否在合理范围内。 更新频率： 历史数据一般是T+1更新。如果你需要实时的，那是另外的服务。 格式： 通常提供CSV格式，方便用Python的pandas或者R直接读取。也有直接对接数据库的API方式，适合集成到自动化系统里。 找到一份规整、可靠的数据能省下很多前期清洗和整理的功夫，让你更专注于策略逻辑本身。不过也要记住，没有完美的数据，了解你所用数据的来源、处理方法和潜在缺陷，和设计策略本身一样重要。 好了，关于数据字段和内容就先聊这么多。这东西写起来比想象中枯燥，希望能帮到正在找数据的你。如果有什么字段没讲清楚，或者你用的时候发现了什么有意思的细节，欢迎留言聊聊。 在充满不确定性的市场中，能够穿越牛熊、实现长期稳定复利的策略是每位宽客的终极追求。今天，我们将为大家深度揭秘一个经过10年超长周期验证、累计收益高达 3815.57%（38倍）的硬核量化策略。 该策略不仅在收益上具备极强的爆发力，更在防御端做到了极致——长达十年的时间跨度内，最大回撤仅 22.21%。它是如何做到的？我们来看看它的核心亮点与底层逻辑。 ? 核心数据亮点 惊人的收益复利 累计收益：3815.57% 年化收益：45.84% 十年时间实现资产跨越式增长。无论是牛市的主升浪，还是震荡市的局部行情，策略都把握得斩钉截铁。 无惧牛熊的极佳防守（核心优势） 最大回撤：仅 22.21%（且回撤区间在2017-2018年的极端熊市） 索提诺比率：2.402（惩罚下行波动的指标，远大于1，说明下行风险极小） 相比基准收益仅24.09%，策略的超额收益高达3055%，这意味着收益并非靠“死扛”大盘Beta，而是刀刀见血的Alpha（Alpha值高达0.428）。 卓越的盈亏性价比（不对称优势） 盈亏比：2.35（平均每次赚钱的幅度是亏钱的2.3倍） 胜率：53.9% (盈利次数642次，亏损548次) 策略不过度追求不切实际的极限胜率，而是通过“截断亏损，让利润奔跑”，实现了极高的容错率。高达 1.719 的夏普比率证明了其极高的单位风险回报。 ⚙️ 策略底层逻辑揭秘：为什么它能稳赚10年？ 通过深度解析策略的底层架构，我们总结出其持续盈利的四大核心武器： 灵敏的“大小盘择时”与“空仓避险”机制（顺势而为） 策略并没有死守某一种风格，而是通过对比代表大盘（沪深300）与小盘（399101）前10日的涨跌动能（Momentum）进行动态切换： 小盘占优时：果断切入高爆发的小市值股票； 大盘占优时：切换至白马/蓝筹阵营避险； 极端泥沙俱下时（大小盘均下跌）：策略直接触发**“持币观望”**机制。空仓是规避系统性风险最强大的武器，这也是为什么策略能在历次股灾中保全本金的核心原因。 独创“市场冷暖温度计”，自适应选股 在切换到大盘股（白马股）时，策略引入了独创的市场温度感知模型（Market Temperature）。通过衡量当前大盘处于近一年来的相对高低位，将市场定义为“Cold（冷）”、“Warm（暖）”和“Hot（热）”： 冷冬市场：聚焦极端低估值（PB<1）、高分红潜力的深度价值股； 温和市场：兼顾价值与成长，选取经营现金流与利润均向好的复苏型企业； 过热市场：拥抱高景气度、高成长性（净利润年增>20%）的强势龙头。 这种动态切换避免了传统价值投资的“估值陷阱”，也避开了成长投资的高位接盘。 符合最新“国九条”的高质量微盘股筛选 在小市值选股上，该策略提前迎合了最苛刻的监管与价值要求。并不是无脑炒垃圾股，而是： 严格过滤 ST、新股、停牌股； 硬核财务底线：归母净利润必须>0，且营业总收入 > 1亿！精准清除了潜在的微盘“退市雷”和保壳垃圾股； 在高财务安全垫的基础上，通过 ROE（净资产收益率）和 ROA（总资产收益率）双剑合璧选出真正的“小而美”。 铁血的日内“双重止损防线”（拒绝死扛） 高盈亏比的根源在于其铁血的除草机制： 个股 8% 硬止损：盘中 14:30 和 14:50 分两次检测，跌破成本价8%（平均成本*0.92）立刻无条件清仓，绝不允许单只股票造成重创。 打板监控：对前一日涨停的持仓股进行特别关照。如果次日尾盘无法继续封死涨停，果断获利了结，有效防止了天地板和游资收割。 概述 在搭建外汇实盘策略、行情监控工具与离线回测框架时，动态增减货币对是高频工程需求。若订阅更新逻辑设计不完善，极易产生行情断档、冗余 Tick 堆积、指标计算延迟抬升等问题，直接干扰交易信号输出，同时造成回测样本失真。本文结合多轮实盘调试与回测校验经验，梳理动态订阅场景下的数据同步痛点，提供分层解耦、增量更新的标准化实现思路，配套可直接集成至量化项目的 Python 代码。 一、量化体系下的动态订阅应用场景 量化开发中存在两类必须动态调整观测标的典型场景： 交互式行情观测系统：根据波动率、流动性人工筛选货币对，短时间内频繁调整订阅清单，多用于盘中机会扫描、多品种对比研究； 自动化轮动策略引擎：模型依据预设阈值自动切换标的池，批量轮换观测品种，适合多因子轮动、跨品种套利类策略。 多数开发者初期会采用「断开连接、全量重订阅」的简易实现，该方式仅适用于离线静态测试，无法满足实盘与高精度回测要求。重连间隙会形成数据空白区间，丢失关键盘口价差；频繁建连销毁还会拉高带宽消耗，甚至触发接口访问限流。 量化工程层面的核心约束：维持单一持久 WebSocket 通道、切换过程无数据缺失、降低无效数据流、控制客户端队列负载，保障实盘信号与回测数据集的完整性。 二、订阅切换引发数据异常的底层成因 外汇实时行情 API 采用「单传输链路 + 多标的主题」架构，服务端仅依据客户端指令推送数据，不会自动做订阅变更优化。未做状态管控时，三类问题会持续影响量化数据质量： 全量重置订阅逻辑：每次清空全部标的再批量订阅，切换窗口期无任何 Tick 输入，K 线重构、滑点统计、回测样本全部出现缺失； 仅新增不注销过期标的：长期累积无用品种数据流，本地消息队列持续膨胀，拖慢指标迭代与模型推理速度； 短周期连续下发订阅指令：频繁增减标的打乱服务端推送时序，产生大量重复、滞后的冗余行情，增加数据清洗开销。 通过日志回放与回测复现可定位问题根源：未隔离「当前有效标的集合」与「目标观测标的集合」，缺少集合差集运算、变更防抖缓冲两层基础处理逻辑。 三、适配量化系统的分层优化架构 3.1 双层解耦模块设计 将行情链路拆分为独立的连接管理层、订阅状态管理层，二者互不耦合，从底层消除切换带来的数据流抖动： 连接管理层：仅负责 WebSocket 心跳保活、异常断线自动重连，无论标的池如何调整，传输通道持续保持连接，避免数据断连； 订阅状态管理层：完成新旧标的集合对比、增减指令拆分，所有订阅变更仅在本模块执行，不改动底层传输链路。 3.2 基于集合差集的增量订阅更新 摒弃数组全覆盖的粗暴写法，使用 Set 结构存储当前生效、目标观测两组标的，通过集合运算区分两类操作： 待新增标的 = 目标集合 − 当前生效集合 待注销标的 = 当前生效集合 − 目标集合 分开发送订阅、注销请求，而非一次性重置全部标的，既消除切换空白窗口，也避免长期堆积无效行情数据。在行情采集与回测数据预处理项目中，我采用 AllTick API 作为数据源，统一规范的请求报文结构，可直接代入差集计算结果下发，降低对接调试成本。 3.3 200ms 防抖缓冲合并短时变更 人工筛选、模型自动轮动均会造成短时间内多次调整标的清单。设置 200ms 缓冲窗口，窗口内全部变更统一合并运算，仅执行一次订阅更新，减少接口请求频次，稳定服务端推送节奏。 3.4 本地队列滞后数据过滤机制 下发注销指令后，服务端会短暂推送延迟 Tick。在数据入计算模块前增加过滤逻辑，剔除已注销货币对的滞后行情，避免脏数据干扰技术指标、量化模型的运算结果。 工程设计核心思路 处理动态订阅应采用「状态集合管控思维」，而非逐次处理单次订阅 / 注销指令。持续维护一份准确的有效标的集合，基于差值做增量更新，代码可维护性更强，便于回测复现、线上故障排查。整套优化的核心目标，并非单纯实现标的切换功能，而是保障数据流在变更过程中完整连续，维持实盘与回测数据的一致性。 四、量化项目可直接集成代码示例 import json import websocket from typing import Set # 当前生效订阅货币对 current_symbols: Set = {"EURUSD", "GBPUSD"} # 切换后的目标观测标的池 target_symbols: Set = {"EURUSD", "USDJPY", "AUDUSD"} # 集合差集计算增减清单 add_list = list(target_symbols - current_symbols) remove_list = list(current_symbols - target_symbols) def refresh_subscription(ws_conn): global current_symbols # 下发新增订阅请求 if add_list: sub_cmd = json.dumps({"action": "subscribe", "params": add_list}) ws_conn.send(sub_cmd) # 下发注销订阅请求 if remove_list: unsub_cmd = json.dumps({"action": "unsubscribe", "params": remove_list}) ws_conn.send(unsub_cmd) # 同步本地有效标的状态 current_symbols = target_symbols.copy() 五、落地适配说明 这套分层增量订阅架构轻量化、无额外算力开销，可适配本地回测采集程序、云部署自动化策略、多品种监控面板等各类量化场景。通过持久连接、变更合并、滞后数据过滤三层优化，解决数据断层、冗余流量、高频请求三类影响数据可信度的核心问题，统一实盘与回测的行情采集逻辑，减少因数据处理差异带来的策略偏差。 
