【Python量化系统工程实战 #05】多策略并行怎么不打架？进程隔离+配置化策略工厂

一个策略跑得好不代表三个策略一起跑也能好。日志互相覆盖、数据互相污染、一个策略崩溃把整组进程全带崩——本篇用 Python 的 multiprocessing 策略工厂 独立日志把「多策略并行」做成干净可扩展的工程模块。一、为什么多策略并行会「打架」很多开发者从单策略过渡到多策略时会直接用同一个 Python 进程顺序执行strategy_a.run()strategy_b.run()strategy_c.run()问题是日志混成一团三个策略的 print 交错输出排查时根本分不清状态互相污染策略 A 改了全局变量策略 B 莫名其妙出错一个崩溃全完蛋策略 A 网络超时抛异常策略 B/C 还没跑就被迫退出没法水平扩展顺序执行耗时 A B C无法利用多核解决思路每个策略一个独立进程 统一配置 独立日志 结果回传。二、本文你将得到什么策略工厂模式新增策略只需注册不改主流程代码multiprocessing 进程隔离每个策略独立 PID、独立内存、互不影响独立日志体系按策略分文件排查时直奔主题统一配置注册表策略参数外置调整无需改代码三、策略工厂新增策略只需一行注册把策略实现和策略调度解耦importpandasaspddefstrategy_sma(df,short5,long20):SMA 金叉死叉策略dfdf.copy()df[sma_short]df[c].rolling(short).mean()df[sma_long]df[c].rolling(long).mean()df[signal]0df.loc[df[sma_short]df[sma_long],signal]1df.loc[df[sma_short]df[sma_long],signal]-1tradesdf[df[signal]!df[signal].shift(1).fillna(0)]return{name:SMA_Cross,last_signal:int(df[signal].iloc[-1]),trade_count:len(trades),}defstrategy_rsi(df,period14,oversold30):RSI 超卖反弹策略dfdf.copy()deltadf[c].diff()gaindelta.where(delta0,0.0)loss(-delta).where(delta0,0.0)rsgain.rolling(period).mean()/loss.rolling(period).mean().replace(0,1e-9)df[rsi]100-(100/(1rs))triggersdf[df[rsi]oversold]return{name:RSI_Bounce,last_rsi:round(float(df[rsi].iloc[-1]),2),trigger_count:len(triggers),}# 策略注册表新增策略只需在这里加一行STRATEGIES{sma:strategy_sma,rsi:strategy_rsi,}核心好处主调度代码只认STRATEGIES注册表新增策略时完全不用改调度逻辑。四、multiprocessing 进程隔离各跑各的用multiprocessing.Process给每个策略分配独立进程崩溃也不互相影响importmultiprocessingasmpimportloggingdefrun_strategy(name,code,params,log_path,result_queue):# 每个进程独立日志logging.basicConfig(levellogging.INFO,format%(asctime)s [%(name)s] %(message)s,handlers[logging.FileHandler(log_path,encodingutf-8),logging.StreamHandler(),],)loglogging.getLogger(name)log.info(f启动 strategy{name}code{code})# 拉取数据内部实现省略dffetch_kline(code)funcSTRATEGIES.get(name)resultfunc(df,**params)result[ok]Truelog.info(f完成 result{result})result_queue.put(result)主调度入口config[{name:sma,params:{short:5,long:20},log:strategy_sma.log},{name:rsi,params:{period:14,oversold:30},log:strategy_rsi.log},]result_queuemp.Queue()processes[]forcfginconfig:pmp.Process(targetrun_strategy,args(cfg[name],600519,cfg[params],cfg[log],result_queue))p.start()processes.append(p)# 等待全部完成forpinprocesses:p.join(timeout60)# 收集结果results[]whilenotresult_queue.empty():results.append(result_queue.get())实测输出[start] sma - pid9476, logstrategy_sma.log [start] rsi - pid3388, logstrategy_rsi.log [ok] RSI_Bounce: last_rsi50.79, trigger_count0 [ok] SMA_Cross: last_signal1, trade_count1 strategy_sma.log: 488 bytes, 3 lines strategy_rsi.log: 407 bytes, 3 lines两个策略并行运行PID 完全独立日志分文件存储。五、统一配置注册表参数外置把策略列表和参数写进 JSON 配置主代码只负责读取和执行{strategies:[{name:sma,params:{short:5,long:20},log:strategy_sma.log},{name:rsi,params:{period:14,oversold:30},log:strategy_rsi.log}],stocks:[600519,000001],risk_limits:{max_drawdown:-0.05}}主代码加载importjsonwithopen(config.json,r,encodingutf-8)asf:cfgjson.load(f)forstrategy_cfgincfg[strategies]:# 直接按配置启动无需硬编码print(f注册策略:{strategy_cfg[name]}params{strategy_cfg[params]})好处调参不用改 Python 代码改 JSON 即可Git 历史清晰能追踪「哪次参数变更带来了收益变化」。六、常见坑Queue 没消费完就退出主进程join()后一定要把所有结果从 Queue 取出否则子进程可能处于僵尸状态。日志Handler重复追加如果在循环里多次logging.basicConfig()同一个进程的日志会被重复打印。确保每个进程只配置一次。Windows 下 multiprocessing 必须加if __name__ __main__否则子进程会递归导入主模块导致无限启动。共享 DataFrame 的陷阱不要试图把大 DataFrame 通过 Queue 传来传去要么每个进程自己拉数据要么用共享内存multiprocessing.shared_memoryPython 3.8。七、小结本文把「多策略并行」从「一个脚本里顺序跑」升级为「工厂 进程隔离 配置驱动」的工程架构策略工厂新增策略只需注册主流程零改动进程隔离每个策略独立 PID、独立日志、独立崩溃域配置外置参数放 JSON版本可控、热更新友好结果回传Queue 收集各策略结果主进程统一汇总下一篇#06我们将推进到「从本地脚本到云端部署」——给你一个量化系统的最小可运行架构模板拿到就能用。免责声明本文仅供技术学习交流不构成任何投资建议。量化策略回测表现不代表未来收益投资有风险决策需谨慎。代码与文档https://github.com/MaiRuiApi