请大家不要客气，任何意见建议可以在这里评论提出。 被采纳后我们将奖励1G研究环境内存 3个月。 调用微信推送报错 发送消息通知失败: HTTPConnectionPool(host='wxpusher.zjiecode.com', port=80): Max retries exceeded with url: /api/send/message (Caused by ConnectTimeoutError(urllib3.connection.httpconnection, 'Connection to wxpusher.zjiecode.com timed out. (connect timeout=6)')) 通过webhook调用钉钉推送也失败， 是supermind内部网络问题？ 大盘到底强不强？用 Python 做一个市场温度计 很多人判断市场强弱，习惯看指数：上证指数涨了，觉得行情不错；创业板跌了，觉得市场很差。 但指数有一个问题：它是少数权重股共同影响的结果。指数上涨，不代表大多数股票都在涨；指数下跌，也不代表没有结构性机会。 更好的办法是看市场广度，也就是回答几个更细的问题： 全市场有多少股票上涨？ 有多少股票涨幅超过 3%？ 有多少股票跌幅超过 3%？ 多少股票站上 20 日均线？ 成交额集中在少数股票，还是扩散到更多标的？ 这篇文章用 AlphaFeed 写一个“市场温度计”：拉取全市场实时行情，再结合历史 K 线计算广度指标，输出一个简单但很有用的市场状态报告。 1. 为什么只看指数不够 假设某一天沪深 300 涨了 1%，你可能会觉得市场很强。 但真实情况可能有三种： 情况 市场含义 大多数股票上涨，指数也上涨 普涨，赚钱效应较好 少数权重股上涨，很多小票下跌 指数强，个股弱 指数小涨，但涨停家数很多 结构性行情活跃 如果只看指数，这三种情况会被混在一起。 所以做交易和研究时，我更喜欢看“市场内部结构”：上涨家数、下跌家数、强势股比例、弱势股比例、均线以上比例、成交额分布。 这些指标不神秘，Python 几十行代码就能算出来。关键是你需要一个能方便拿到全市场行情的数据源。 2. 用 AlphaFeed 获取全市场实时行情 安装： pip install alphafeed pandas 初始化： from alphafeed import AlphaFeed af = AlphaFeed(api_key="your-api-key") 也可以用环境变量： export ALPHAFEED_API_KEY="your-api-key" 然后代码里直接： from alphafeed import AlphaFeed af = AlphaFeed() 获取全部 A 股实时行情： import pandas as pd from alphafeed import AlphaFeed af = AlphaFeed() quotes = af.quotes.get(universes=["CN_Stock"], to_dataframe=True) print(len(quotes)) print(quotes[["symbol", "last_price", "prev_close", "volume", "amount", "ext.name", "ext.change_pct"]].head()) AlphaFeed 支持的标的池包括： 标的池 含义 CN_Stock A 股 CN_ETF ETF US_Stock 美股 HK_Stock 港股 这意味着你可以用同样的方法做 A 股市场温度计、ETF 市场温度计、美股观察面板或港股观察面板。 3. 计算最基础的市场广度 先计算上涨、下跌、平盘数量： def calc_basic_breadth(quotes): df = quotes.copy() change_pct = df["ext.change_pct"] total = len(df) up = int((change_pct > 0).sum()) down = int((change_pct < 0).sum()) flat = total - up - down return { "total": total, "up": up, "down": down, "flat": flat, "up_ratio": up / total if total else 0, "down_ratio": down / total if total else 0, } 打印结果： breadth = calc_basic_breadth(quotes) print(f"全市场标的数: {breadth['total']}") print(f"上涨家数: {breadth['up']} ({breadth['up_ratio']:.1%})") print(f"下跌家数: {breadth['down']} ({breadth['down_ratio']:.1%})") print(f"平盘家数: {breadth['flat']}") 这个指标很简单，但已经比只看指数多了一层信息。 比如指数涨 0.5%，但上涨家数只有 25%，那说明行情可能只是权重股在撑指数。反过来，指数跌 0.3%，但上涨家数超过 60%，说明市场内部可能并不弱。 4. 计算强势股和弱势股比例 再定义几个阈值： def calc_strength_distribution(quotes): df = quotes.copy() change_pct = df["ext.change_pct"] return { "rise_gt_3pct": int((change_pct >= 0.03).sum()), "rise_gt_5pct": int((change_pct >= 0.05).sum()), "fall_gt_3pct": int((change_pct <= -0.03).sum()), "fall_gt_5pct": int((change_pct <= -0.05).sum()), } 这些指标可以粗略判断赚钱效应： 指标 含义 涨幅超 3% 家数 主动进攻力量 涨幅超 5% 家数 强势股扩散程度 跌幅超 3% 家数 风险释放范围 跌幅超 5% 家数 恐慌程度 很多时候，大盘指数涨跌不大，但强势股数量会提前变化。市场温度计的价值就在这里：它不是预测明天，而是更客观地描述今天。 5. 找出成交额最高的股票 成交额代表市场注意力。我们可以看资金主要集中在哪里： top_amount = quotes.sort_values("amount", ascending=False).head(20) print( top_amount[[ "symbol", "ext.name", "last_price", "ext.change_pct", "amount", ]] ) 如果每天成交额前 20 名都是少数权重股，说明市场可能偏指数行情；如果成交额扩散到很多中小市值股票，说明市场活跃度可能更高。 当然，AlphaFeed 这里只提供行情数据，不替你判断行业主题。你可以把成交额榜导出后，再结合自己的行业分类或观察列表做进一步分析。 6. 加入 20 日均线以上比例 实时行情能告诉我们今天涨跌，但判断趋势还需要历史 K 线。 我们可以抽取成交额较高的一批股票，计算有多少站上 20 日均线： active_symbols = ( quotes.sort_values("amount", ascending=False) .head(300)["symbol"] .tolist() ) dfs = af.klines.batch( active_symbols, period="1d", count=60, adjust="forward", to_dataframe=True, show_progress=True, ) 计算比例： def calc_above_ma20_ratio(dfs): total = 0 above = 0 for symbol, df in dfs.items(): df = df.sort_values("trade_date").copy() if len(df) < 20: continue ma20 = df["close"].rolling(20).mean().iloc[-1] close = df["close"].iloc[-1] if pd.isna(ma20): continue total += 1 if close > ma20: above += 1 return above / total if total else 0 使用： above_ma20_ratio = calc_above_ma20_ratio(dfs) print(f"活跃股票中站上20日线比例: {above_ma20_ratio:.1%}") 这个指标很适合做市场温度判断： 20 日线以上比例 粗略状态 70% 以上 市场较热 50% - 70% 偏强或震荡偏强 30% - 50% 震荡偏弱 30% 以下 市场较冷 阈值只是经验，不要机械使用。不同市场、不同时期都需要校准。 7. 输出一份市场温度报告 把上面的部分合起来： import pandas as pd from datetime import datetime from alphafeed import AlphaFeed def calc_basic_breadth(quotes): change_pct = quotes["ext.change_pct"] total = len(quotes) up = int((change_pct > 0).sum()) down = int((change_pct < 0).sum()) flat = total - up - down return total, up, down, flat def calc_strength_distribution(quotes): change_pct = quotes["ext.change_pct"] return { "rise_gt_3pct": int((change_pct >= 0.03).sum()), "rise_gt_5pct": int((change_pct >= 0.05).sum()), "fall_gt_3pct": int((change_pct <= -0.03).sum()), "fall_gt_5pct": int((change_pct <= -0.05).sum()), } def calc_above_ma20_ratio(dfs): total = 0 above = 0 for _, df in dfs.items(): df = df.sort_values("trade_date").copy() if len(df) < 20: continue ma20 = df["close"].rolling(20).mean().iloc[-1] close = df["close"].iloc[-1] if pd.isna(ma20): continue total += 1 above += int(close > ma20) return above / total if total else 0 def market_temperature_label(up_ratio, above_ma20_ratio, weak_count): if up_ratio >= 0.65 and above_ma20_ratio >= 0.6: return "偏热" if up_ratio >= 0.5 and above_ma20_ratio >= 0.45: return "偏强" if up_ratio <= 0.35 or weak_count > 500: return "偏冷" return "震荡" def build_market_report(universe="CN_Stock"): af = AlphaFeed() quotes = af.quotes.get(universes=[universe], to_dataframe=True) quotes = quotes.dropna(subset=["ext.change_pct", "amount"]) total, up, down, flat = calc_basic_breadth(quotes) dist = calc_strength_distribution(quotes) active_symbols = ( quotes.sort_values("amount", ascending=False) .head(300)["symbol"] .tolist() ) dfs = af.klines.batch( active_symbols, period="1d", count=60, adjust="forward", to_dataframe=True, show_progress=True, ) above_ma20_ratio = calc_above_ma20_ratio(dfs) up_ratio = up / total if total else 0 label = market_temperature_label( up_ratio, above_ma20_ratio, dist["fall_gt_3pct"], ) top_amount = quotes.sort_values("amount", ascending=False).head(10) lines = [] lines.append(f"# 市场温度计 {datetime.now():%Y-%m-%d %H:%M}") lines.append("") lines.append(f"- 标的池：{universe}") lines.append(f"- 市场状态：{label}") lines.append(f"- 上涨家数：{up} / {total} ({up_ratio:.1%})") lines.append(f"- 下跌家数：{down} / {total} ({down / total:.1%})") lines.append(f"- 平盘家数：{flat}") lines.append(f"- 涨幅超过 3%：{dist['rise_gt_3pct']}") lines.append(f"- 涨幅超过 5%：{dist['rise_gt_5pct']}") lines.append(f"- 跌幅超过 3%：{dist['fall_gt_3pct']}") lines.append(f"- 跌幅超过 5%：{dist['fall_gt_5pct']}") lines.append(f"- 活跃股票站上 20 日线比例：{above_ma20_ratio:.1%}") lines.append("") lines.append("## 成交额 Top 10") lines.append("") lines.append(top_amount[["symbol", "ext.name", "last_price", "ext.change_pct", "amount"]].to_markdown(index=False)) lines.append("") lines.append("数据来源：AlphaFeed (https://alphafeed.org/)") return "\n".join(lines) if __name__ == "__main__": report = build_market_report("CN_Stock") print(report) with open("market-temperature.md", "w", encoding="utf-8") as f: f.write(report) 如果需要 to_markdown，安装： pip install tabulate 8. 这个报告怎么读 你可以每天收盘后生成一次报告，然后观察几个核心变化： 指标 观察重点 上涨家数比例 市场赚钱效应是否扩散 跌幅超过 3% 家数 风险释放是否扩大 活跃股站上 20 日线比例 趋势是否健康 成交额 Top 10 资金是否集中在少数权重股 市场状态标签 给自己一个快速摘要 不要把它当成买卖信号。它更像天气预报里的温度和湿度：告诉你环境如何，但不会替你决定今天穿哪件衣服。 交易决策仍然要结合你的策略、仓位、风险承受能力和交易计划。 9. 扩展到 ETF、美股、港股 如果你想看 ETF 市场： report = build_market_report("CN_ETF") 如果你想看美股： report = build_market_report("US_Stock") 如果你想看港股： report = build_market_report("HK_Stock") 跨市场统一接口的好处在这里会非常明显。你的市场温度计逻辑不用重写，只要换一个标的池 ID，就可以观察不同市场。 当然，不同市场的交易制度、涨跌幅分布、成交结构都不一样，所以阈值不要照搬。比如 A 股和美股的单日波动结构不同，涨幅超过 3% 在不同市场里的含义也不同。 10. 可以继续做成一个 Dashboard 这篇文章输出的是 Markdown 报告。如果你想更进一步，可以做成网页看板： 用 Streamlit 显示市场温度； 用 Plotly 画上涨/下跌比例柱状图； 每 5 分钟刷新一次实时行情； 保存每日市场温度到 CSV； 画出市场温度历史曲线； 给极端冷/热状态加提醒。 你甚至可以让 AI 编程工具帮你改： 请基于这个 AlphaFeed 市场温度计脚本，帮我改成 Streamlit Dashboard： 1. 页面顶部显示市场状态标签； 2. 显示上涨/下跌/平盘数量； 3. 显示涨幅超过 3%、跌幅超过 3% 的数量； 4. 显示成交额 Top 20 表格； 5. 支持在侧边栏选择 CN_Stock、CN_ETF、US_Stock、HK_Stock。 这种需求非常适合 AlphaFeed + AI 编程：AlphaFeed 提供结构化数据，AI 帮你快速搭界面，你把注意力放在指标设计和结果解释上。 11. 为什么这是一个适合普通人的量化工具 很多人一听量化，就觉得一定要机器学习、深度学习、高频交易。其实不是。 对大多数个人投资者和研究者来说，更实用的量化工具往往是： 自动整理市场信息； 用固定规则减少主观情绪； 长期记录市场状态； 用数据验证自己的感觉； 在风险变大时及时降温。 市场温度计就是这样的工具。它不复杂，但每天都能用。 当你觉得“今天市场好像很强”时，可以看上涨家数比例；当你觉得“指数没跌多少，但账户很难受”时，可以看跌幅超过 3% 的家数；当你觉得“是不是该加仓”时，可以先看活跃股票站上 20 日线的比例。 它不能替你赚钱，但能帮你少一点拍脑袋。 结语 判断市场强弱，不应该只盯着指数涨跌。指数是结果，市场广度才更接近交易者真实体感。 用 AlphaFeed，你可以用几行 Python 拿到全市场实时行情，再结合历史 K 线计算自己的市场温度指标。这个过程没有玄学，也不需要复杂模型，核心就是把“感觉”变成“可重复观察的数据”。 如果你正在搭建自己的量化工作流，可以从这个市场温度计开始：每天生成一份报告，持续记录，慢慢你会对市场状态有更稳定的判断。 参考文献： AlphaFeed 官网：https://alphafeed.org/ AlphaFeed 文档：https://docs.alphafeed.org/ CMES金融数据库能下载哪些期货数据？一次讲清楚 昨晚跑策略回测，又差点把内存撑爆了。复盘了一下，问题出在数据源上。之前图省事，用的数据要么字段不全，要么不是标准清洗过的，自己处理起来费时费力还容易出错。 后来开始用CMES金融数据库，主要是因为它数据分类比较清楚，下载也方便。今天就把我研究过的、它上面能获取的几大类数据整理一下，给同样做量化或者需要数据的朋友参考。内容比较干，可以先收藏。 商品期货 & 金融期货 这个是最基础的，做期货策略肯定绕不开。数据主要包括日线、分钟线这些。 日线数据：就是每天一根K线，有日期、开盘价、最高价、最低价、收盘价、成交量、持仓量这些经典字段。适合做中长线策略的初步回测，或者宏观分析。 分钟线数据：频率更高，比如1分钟、5分钟、15分钟线。字段和日线差不多，但时间粒度更细。做日内或者中短线策略，这个数据基本是必需品。我刚开始的时候，傻傻分不清分钟线和Tick，用Tick数据去做日频回测，那效率简直了，跑一个策略等半天。 Level 2 行情数据（五档订单簿） 这东西数据量就上来了，也是让我又爱又恨的部分。爱的是信息量足，恨的是真的占地方。 它记录的是市场上实时的买卖委托订单，通常展示买卖各五个价位的挂单情况。核心字段包括： 时间戳：精确到毫秒，这是做高频或微观结构分析的基础。 最新价、成交量：这个和普通行情一样。 买一价到买五价，买一量到买五量：就是当前市场上最想买的五个价格和对应的挂单数量。 卖一价到卖五价，卖一量到卖五量：同理，最想卖的五个档位。 总买委托量、总卖委托量：有时候看这个能感受一下多空力量的对比。 以前我只看K线，后来发现Level 2里的信息量完全不同。比如，你可以观察委托队列的动态变化，有时候买一挂了个巨量单子，价格却上不去，可能那是个“假墙”，是拆单拆出来的。为了验证一些盘口规律，我调取了CMES金融数据库中过去三年的主力合约五档数据进行回测，发现结合委托信息，确实能过滤掉不少市场噪音。 Tick数据（一档） 如果说分钟线是总结报告，那Tick数据就是现场直播。市场每发生一笔成交，或者报价有变动，它就记录一条。数据频率极高，是研究市场微观结构、做高频交易策略的底层材料。 主要字段通常包括： 时间戳：同样精确到毫秒或更高。 最新价 成交量 成交额 买一价、买一量 卖一价、卖一量 Tick数据是硬盘杀手，新手真的不建议一上来就碰，数据清洗和存储都是大工程。我之前用自己爬的Raw Tick数据，光是处理异常值和合并合约就搞到怀疑人生。 更高频的数据（如逐笔委托/成交） 这个在CMES上好像也有，属于更深的Level。比如每一笔委托的到达、每一笔成交的明细。数据量是Tick的N倍，对存储和算力要求极高，一般是专业机构玩得多。个人研究者用五档数据其实已经能挖不少东西了。 不同数据的简单对比 用文字描述可能更直观一些： Tick数据：像个碎碎念的监控器，市场动一下它就记一笔，信息最原始，数据量最大，存储和处理成本高。 分钟/小时数据：像个定时的记录员，每隔固定时间给你汇报一次市场概况，数据量适中，适合大多数技术分析和中短线回测。 日/周/月数据：像最终的日报、周报、月报，高度概括，数据量小，适合长期趋势分析、基本面量化入门。 怎么获取和使用？ 他们网站提供了下载界面，可以根据品种、日期、数据类型筛选。对于需要编程获取的情况，他们好像也提供了API接口，用Python就能调。我之前试过，挺方便的。 这里放一个他们API文档里示例的代码片段（具体参数以最新文档为准）： # 示例：使用cmesdata库获取数据 # 注意：使用前需要pip安装，并确保拥有有效的访问权限和正确的参数 import cmesdata as cd # 初始化客户端，需要替换为你的实际token client = cd.Client(api_token='your_api_token_here') # 尝试获取某商品期货的分钟线数据 # CMES金融数据库的行情接口，注意入参正确，调用频率正常。 try: data = client.get_data( symbol='RB9999', # 合约代码，例如螺纹钢主力 data_type='1min', # 数据类型：1分钟线 start_date='2023-01-01', end_date='2023-01-10' ) print(data.head()) except Exception as e: print(f"获取数据失败: {e}") 数据是量化研究的基石，但找到干净、规整、字段统一的数据源真的能省下大把时间。我一开始也到处找免费数据拼凑，结果在数据清洗和校验上花的功夫比研究策略还多。后来还是转向了CMES金融数据库这类提供标准化清洗数据的地方，虽然需要一些积分或者成本，但数据质量有保障，不用自己处理除权除息、合约换月这些破事，省下的时间可以多琢磨几个策略逻辑。 当然，数据只是工具，关键还是看你的研究思路。不建议一开始就追求最高频的数据，从日线、分钟线入手，把策略逻辑跑通，再逐步用到更细粒度的数据，可能会更稳妥。 大概就这些吧，手都打酸了。如果大家对这些数据的具体字段还有疑问，或者知道怎么高效压缩存储Tick数据。 JST1:=HHV(H,3); JST2:=H>REF(JST1,1); JST3:=REFXV(JST1,3); JST4:=H>=JST3; JST5:=JST4 AND JST2; JST6:=LLV(L,3); JST7:=L<REF(JST6,1); JST8:=REFXV(JST6,3); JST9:=L<=JST8; JST10:=JST9 AND JST7; JST11:=IF(JST5 AND JST10,TOPRANGE(H)>LOWRANGE(LOW),JST5); JST12:=IF(JST5 AND JST10,TOPRANGE(H)<LOWRANGE(LOW),JST10); JST13:=1; JST14:=BARSLAST(JST11); JST15:=BARSLAST(JST12); JST16:=COUNT(JST11,JST15+1)=1 AND JST11; JST17:=COUNT(JST12,JST14+1)=1 AND JST12; JST18:=BARSLAST(JST16); JST19:=BARSLAST(JST17); JST20:=BACKSET(REFX(JST17,1),HHVBARS(IF(JST11,H,0),REF(JST15,1)+2)+1); JST21:=BACKSET((BARSTATUS=2 AND JST19>JST18),HHVBARS(IF(JST11,H,0),REF(JST15,1)+2)+1); JST22:=JST20 OR JST21; JST23:=COUNT(JST22,2)=1 AND JST22; JST24:=BACKSET(REFX(JST16,1),LLVBARS(IF(JST12,L,500000),REF(JST14,1)+2)+1); JST25:=BACKSET((BARSTATUS=2 AND JST19<JST18),LLVBARS(IF(JST12,L,500000),REF(JST14,1)+2)+1); JST26:=JST24 OR JST25; JST27:=COUNT(JST26,2)=1 AND JST26; JST28:=(JST23 AND H>REF(L,BARSLAST(JST27))); JST29:=(JST27 AND L<REF(H,BARSLAST(JST23))); JST30:=BARSLAST(JST28); JST31:=BARSLAST(JST29); JST32:=COUNT(JST28,JST31+1)=1 AND JST28; JST33:=COUNT(JST29,JST30+1)=1 AND JST29; JST34:=BARSLAST(JST32); JST35:=BARSLAST(JST33); JST36:=BACKSET(REFX(JST33,1),HHVBARS(IF(JST28,H,0),REF(JST31,1)+2)+1); JST37:=BACKSET((BARSTATUS=2 AND JST35>JST34),HHVBARS(IF(JST28,H,0),REF(JST31,1)+2)+1); JST38:=JST36 OR JST37; JST39:=COUNT(JST38,2)=1 AND JST38; JST40:=BACKSET(REFX(JST32,1),LLVBARS(IF(JST29,L,500000),REF(JST30,1)+2)+1); JST41:=BACKSET((BARSTATUS=2 AND JST35<JST34),LLVBARS(IF(JST29,L,500000),REF(JST30,1)+2)+1); JST42:=JST40 OR JST41; JST43:=COUNT(JST42,2)=1 AND JST42; JST44:=IF(JST39=0,0,JST39) AND JST13; JST45:=IF(JST43=0,0,JST43) AND JST13; JST46:=(JST44); JST47:=(JST45); JSTS:=EMA(H-L,300)/2; DRAWLINE(JST46,H,JST47,L,0),LINETHICK1,COLORGREEN; DRAWLINE(JST47,L,JST46,H,0),LINETHICK1,COLORYELLOW; DRAWNUMBER(JST46,H+EMA(H-L,200)/2,H),COLORRED; DRAWNUMBER(JST47,L,L),COLORMAGENTA; STICKLINE(1,REF(H,BARSLAST(JST46)),REF(H,BARSLAST(JST46)),4,0),COLORRED; STICKLINE(1,REF(L,BARSLAST(JST47)),REF(L,BARSLAST(JST47)),4,0),COLORMAGENTA; 我的一大堆文件就这样消失了？？？？重新登陆也不行！ -- 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 
