炒股智能体开发实战：数据清洗、信号合成与回测验证

老实说大多数想自己动手做炒股智能体的人第一步就搞错了方向他们以为核心是“模型得多聪明”结果把精力全砸在调参和换算法上。我做完一版能本地稳定运行的演示系统之后最大的感受是——真正决定这套体系能不能用的是数据链路、信号定义和回测口径。本文从零讲清楚一个炒股智能体是怎么搭出来的行情数据怎么接、怎么洗技术指标怎么变成信号信号怎么过回测以及从回测到实盘辅助之间还缺哪些工程细节。适合有一定 Python 基础、想用程序化方式研究股票行情、又不想被量化框架捆绑的朋友参考。整套东西定位是研究工具不是荐股工具文中所有标的均为演示用虚拟代码不构成任何投资建议。1. 炒股智能体的整体框架与设计底线1.1 智能体不是预测器而是一条信号流水线很多人听到“智能体”三个字就默认它得会预测明天涨跌。我自己第一版也是这么设计的结果发现预测模型的胜率就算做到 55%在实盘里依然很难稳定落地因为中间隔了太多环节数据对不对、信号触发后人工能不能执行、回撤能不能扛住。后来我把思路换了一下一个炒股智能体本质上是一条“信号流水线”。它从行情数据出发经过指标计算、信号合成、回测验证、风险控制最后输出的是带纪律性的观察结果而不是“包涨”的结论。这个定位转变非常重要它决定了我不会在模型上过度投入而是把更多精力放在减少信号失真上。流水线的每一环都可能引入误差数据源有偏、复权方式选错、指标参数写反、回测时偷看了未来数据、手续费和滑点没算进去。任何一个环节出错后面的结果都是自欺欺人。所以我在设计整体框架时核心原则就是“让每一环可检查、可复现、可单独替换”。1.2 四层架构数据、计算、信号、执行辅助整个系统我拆成四层数据层负责拉取历史行情、清洗、入库、做复权处理也负责提供统一的数据读取接口。计算层基于干净的K线数据计算各类技术指标比如均线、MACD、RSI、KDJ。信号层把指标结果组合成具体的交易信号比如“金叉”“超卖反弹”“多指标共振”。执行辅助层负责回测验证、仓位评估、风险提示以及给人工决策生成一份可读的复核清单。为什么不把“自动下单”放进去因为个人研究场景里自动下单涉及账号安全、接口权限、异常处理等多重工程问题稍有不慎就把研究工具变成了风险源头。我的做法是让智能体停在“信号生成 风控提示”这个位置真正下单动作由人工确认完成这既是合规边界也是安全边界。1.3 我对技术栈的取舍少依赖好调试技术选型上我特意保持克制。整体只用了 Python 加 pandas、numpy、sqlite3 这几个基础组件没有引入重量级的量化框架。原因很简单框架越重调试时的黑盒越多。我希望每一行逻辑都是自己能读懂的。数据库用的 SQLite单文件、零部署、够用。历史行情按日线级别存每支演示标的单独一张表字段固定在 date、open、high、low、close、volume 这六项。后续算指标、回测都从这个库里取数确保同一份数据源贯穿全流程不会出现“训练时一套数据、验证时另一套数据”的混乱。这套极简架构还有一个好处方便排查未来函数。因为所有计算逻辑都在自己的代码里一旦回测收益异常高我能直接定位是哪个环节提前用到了未来信息。2. 行情数据的获取、清洗与本地存储2.1 行情数据的字段设计与复权处理做智能体第一步不是写指标而是先把数据底座打牢。我的库表结构设计成这样CREATE TABLE IF NOT EXISTS daily_bar ( symbol TEXT, date TEXT, open REAL, high REAL, low REAL, close REAL, volume INTEGER, PRIMARY KEY (symbol, date) );主键用(symbol, date)可以天然防重复。数据接口统一封装成一个函数外部一律不管数据来源是本地 CSV 还是在线接口只要返回标准字段就行import pandas as pd def fetch_history(symbol: str, start: str, end: str) - pd.DataFrame: 返回标准K线DataFrame必须包含列: date, open, high, low, close, volume # 这里替换成你的数据源客户端或者读取本地CSV raise NotImplementedError(请替换成你自己的数据源实现)需要特别提醒的是复权问题。除权除息会让价格出现跳空如果不做复权处理技术指标的连续性会被破坏。实际使用中我建议用前复权数据做历史回测因为前复权保持了最新价格的真实性信号看起来更接近现在能看到的样子。2.2 数据质量的三个硬性检查我踩过最大的坑就是拿“看起来挺全”的数据直接跑指标结果策略回测曲线漂亮得不像话。后来一查问题出在数据本身。现在我的清洗管线固定做三件事重复检查同一天出现两条K线只保留最后一条。逻辑校验最低价不能大于最高价收盘价必须大于 0成交量不能为负。区间完整性检查交易日是否存在大面积缺失缺失超过一定比例就直接放弃这支标的不硬补。清洗逻辑写成函数就是一二十行的事def clean_bars(df: pd.DataFrame) - pd.DataFrame: df df.drop_duplicates(subsetdate).sort_values(date).reset_index(dropTrue) df df[(df[close] 0) (df[low] df[high]) (df[volume] 0)] return df有些人会问缺口能不能用前值填充我的建议是别。日线级别的缺口经常意味着停牌或数据源本身没返回用插值填出来的K线是“假K线”算出来的指标也是“假信号”。宁可跳过这支标的也不要在脏数据上做研究。2.3 数据获取与入库的示例实现入库逻辑我用to_sql配合if_existsappend再靠主键去重简单可靠import sqlite3 def save_bars(symbol: str, df: pd.DataFrame, db_path: str market.db): df clean_bars(df) df[symbol] symbol conn sqlite3.connect(db_path) df.to_sql(daily_bar, conn, if_existsappend, indexFalse) conn.close()获取数据时顺手把范围参数做好避免每次全量重拉。全量重拉既慢又容易把数据源接口搞出限流问题。3. 核心指标计算与多信号合成逻辑3.1 均线与MACD趋势派的第一梯队数据干净之后开始算指标。技术指标的本质是对历史价格行为的量化描述它不预言未来只是把“过去发生了什么”整理成可比较的数字。均线和 MACD 是趋势识别的基础工具。均线我用简单移动平均直接rolling(window).mean()就能算。MACD 用指数移动平均这里有个细节不能用普通rolling().mean()必须用ewm(span...)才能保持递推特性。标准参数是 12、26、9def calc_macd(close: pd.Series, fast12, slow26, signal9): dif close.ewm(spanfast, adjustFalse).mean() - close.ewm(spanslow, adjustFalse).mean() dea dif.ewm(spansignal, adjustFalse).mean() hist (dif - dea) * 2 return dif, dea, histMACD 的使用上我特别强调一句金叉死叉只是工具不是圣杯。我在 demo 里同时记录 dif 与零轴的位置关系避免在零轴下方反复纠缠时频繁触发信号。3.2 RSI与KDJ摆动指标的钝化与修复RSI 和 KDJ 属于摆动指标适合辅助判断短期超买超卖。RSI 计算时要用 Wilder 平滑它的本质是ewm(alpha1/n)def calc_rsi(close: pd.Series, period14): delta close.diff() gain delta.clip(lower0) loss -delta.clip(upper0) avg_gain gain.ewm(alpha1/period, adjustFalse).mean() avg_loss loss.ewm(alpha1/period, adjustFalse).mean() rs avg_gain / avg_loss rsi 100 - (100 / (1 rs)) return rsiKDJ 的常规算法需要先算 RSV再对 RSV 做指数平滑得到 K 和 DJ 值是三倍 K 减两倍 D。说实话这个指标在震荡行情里表现尚可但在单边行情里会钝化得很明显。所以我的信号层里KDJ 只负责“辅助确认”权重不高。3.3 信号合成的“投票机制”与参数敏感性单指标信号太容易抖动我最后采用“投票机制”做信号合成。每个指标单独输出方向1 表示偏多-1 表示偏空0 表示中性。最后把所有分数加起来按阈值划分信号强度def vote_score(row) - int: score 0 if row[ma_trend] 0: score 1 elif row[ma_trend] 0: score - 1 if row[macd_hist] 0: score 1 elif row[macd_hist] 0: score - 1 if row[rsi] 50: score 1 elif row[rsi] 50: score - 1 if row[kdj_j] 50: score 1 elif row[kdj_j] 50: score - 1 return score投票机制的好处是容错。某个指标在震荡市里反复出错但只要其他几个方向一致最终信号仍能保持稳定。要注意的是参数敏感性把 RSI 周期从 14 改成 7把 MACD 从 12、26 改成 5、35结果会完全不同。做课题研究时最好固定一组参数并做敏感性分析而不是反复往“历史最优”上调。4. 回测引擎与结果验证防止自欺欺人4.1 最小可用回测是怎么算的信号层输出之后不能直接用手感判断好不好必须放到历史数据里走一遍。我的回测引擎非常简单逐日迭代模拟持仓和现金两条线def backtest(bars, signals, initial_cash100000, fee_rate0.0005, slippage0.0005): cash initial_cash position 0 records [] for i in range(len(bars)): date bars[date].iloc[i] close bars[close].iloc[i] sig signals.iloc[i] if sig 1 and position 0: buy_price close * (1 slippage) shares int(cash / buy_price / 100) * 100 cost shares * buy_price * (1 fee_rate) if cost cash: cash - cost position shares records.append((date, BUY, buy_price, shares, cash)) elif sig -1 and position 0: sell_price close * (1 - slippage) cash position * sell_price * (1 - fee_rate) records.append((date, SELL, sell_price, position, cash)) position 0 total_value cash position * close records.append((date, MARK, close, position, total_value)) return pd.DataFrame(records, columns[date, action, price, position, equity])这里的信号阈值用了强信号大于 1 或小于 -1避免高频抖动频繁交易。交易数量按整手 100 股控制更接近真实可执行情况。4.2 手续费、滑点与交易周期假设回测里最容易美化结果的两个地方一个是手续费一个是滑点。很多新手把手续费设成 0滑点设成 0回测曲线当然好看。我建议至少按以下口径估算项目保守估计说明佣金费率0.05%双边收取大资金可以更低滑点0.05%按成交价偏离 5 个基点估算最小交易单位100 股忽略碎股回测更贴近实际还有一个隐含假设是“当天收盘信号当天成交”。实际中信号往往在收盘后确认第二天开盘才能介入。严谨的做法是信号 T 日生成T1 日按开盘价成交。我自己的 demo 里默认 T1 执行虽然会牺牲一点历史收益但换来的结果是“如果真按这个系统操作大概率能重现”。4.3 过拟合识别与参数平替实验回测收益特别高的时候第一反应应该是怀疑而不是高兴。高收益通常意味着某个环节在偷看未来或者参数被无意识优化到了历史行情上。我自己常用的验证手段叫“参数平替实验”把均线从 20 日改成 30 日把 RSI 从 14 改成 21看收益曲线是否依然稳健。如果参数一换收益从 80% 掉到 5%说明策略本身没有稳健的逻辑支撑只是在拟合历史噪声。反之如果收益有波动但整体趋势还在说明策略捕捉到了某种持续存在的市场结构。回测绩效至少看这四个指标总收益率策略在测试区间的整体表现。最大回撤从峰值跌到谷底的最大幅度比收益率更值得关注。胜率盈利交易次数占总交易次数的比例。盈亏比平均盈利除以平均亏损胜率和盈亏比需要一起看。5. 从回测到实盘辅助检查清单与风控模块5.1 上线前我必查的十个问题回测跑通不等于可以直接实盘。我整理了一张检查清单每次准备启用一个新策略前都会过一遍数据源是否稳定有没有停更、乱码、复权错误。信号计算是否存在未来函数T 日信号是否依赖了 T 日的最终收盘价。手续费和滑点是否按保守口径计入。是否有足够的样本外验证只用最近一段行情测出来的结果没有说服力。参数是否做了敏感性分析连续亏损时系统能不能承受最大回撤。单支标的仓位上限有没有设临时停牌、数据缺失时系统如何表现。信号生成后是否需要人工复核复核流程是否清晰。是不是只准备投自己亏得起的钱。这十条里任何一条不过关我都不会把对应策略加入正式观察列表。5.2 风控模块仓位、冷却与信号分级智能体必须内置风控逻辑而不是把风控交给人的临时判断。我在系统里加了三层风控单标的上限单支演示标的占总投资金额不超过 20%防止单一标的风险过大。信号冷却期同一标的在 N 个交易日内只允许执行一次反向操作避免被震荡行情来回打脸。信号分级只有强共振信号才生成高优先级提示单指标信号只作观察记录。代码层面加一个冷却判断也就是记录最近一次交易日期N 天内不再生成新信号def check_hold(signals, last_trade_date, cooldown_days5): if last_trade_date is None: return True days (signals[date] - pd.to_datetime(last_trade_date)).days return days cooldown_days5.3 运行监控与复盘报告系统跑起来以后不能扔在那里不管。我每天收盘后运行一次定时任务把当天信号写入一张 signal_log 表并自动生成一个简短的复盘报告内容包含当日信号、触发依据、组合当前仓位、距离风控阈值还有多远。这种做法最大的价值是留下记录。两周之后再回头看能清楚知道每个信号当时为什么触发也能复盘哪些信号被市场证明是错的。没有记录的智能体和没有日志的程序一样出了问题根本无法定位。6. 踩坑实录我开发过程中最典型的三个问题6.1 前复权数据导致的假突破第一次用前复权数据算突破策略时发现某演示标的频频出现“历史突破信号”回测收益高得离谱。后来排查发现前复权会修正历史价格导致早期价格跟当时真实看到的 K 线完全不同。某些看起来在历史底部发出的买入信号在当时的实时数据里根本不存在。这个问题的教训是回测必须明确标注数据复权口径并且不能在同一个策略里混用前复权和未复权数据。大多数策略我会统一用前复权但在结果分析时要意识到历史信号的形态可能被数据修正过。6.2 信号函数里隐藏的未来函数我有一次回测年化收益做到 300%兴奋之余开始怀疑哪里出了问题。逐行检查后发现指标计算时我把整段历史数据一次性传给了ewm函数但信号判断却在每一天循环里取“当天的最终数据”——这其实是合理的问题出在我用iloc[i1]去判断次日的涨跌作为当天信号本质上把未来信息泄了进去。检查方法很简单把信号日期和收益日期错开一天看收益是否显著下降。如果不降反升说明策略真的依赖当日到次日的价格延续如果收益骤降基本可以确认未来函数存在。建议新手养成习惯信号 DataFrame 和行情 DataFrame 一定用日期做显式对齐不要靠行号隐式对齐。6.3 回测绩效被“幸存者标的”抬高了第三个坑出在选样本上。我最早只在“现在看还活得好好的”标的上回测结果当然是漂亮。但真实场景里历史上有大量标的已经退市或长期低迷如果把它们排除在样本外回测就成了幸存者偏差游戏。修正方案是建立标的池时以某个固定历史时点为基准纳入当时所有满足流动性条件的标的后续无论它涨跌、是否退市都不删除。这个过程需要历史全量数据虽然准备工作麻烦但只有这样才能让回测结果具备真正的参考意义。如果让我重头再做一次我会把更多时间花在数据口径和回测假设上而不是执着于某个指标参数是否更优。炒股智能体的开发难的不是让它变聪明而是让它每一环都诚实、可验证。整套系统现在对我来说更像一个纪律化的观察助手它不替我决策但帮我把情绪和噪声挡在门外。最后再分享一个朴素经验所有听起来完美得不像话的回测曲线背后一定藏着一个没被发现的问题。