

> 一个策略跑得好，不代表三个策略一起跑也能好。日志互相覆盖、数据互相污染、一个策略崩溃把整组进程全带崩——本篇用 Python 的 multiprocessing + 策略工厂 + 独立日志，把「多策略并行」做成干净可扩展的工程模块。

## 一、为什么多策略并行会「打架」

很多开发者从单策略过渡到多策略时，会直接用同一个 Python 进程顺序执行：

```python
strategy_a.run()
strategy_b.run()
strategy_c.run()
```

问题是：

- **日志混成一团**：三个策略的 print 交错输出，排查时根本分不清
- **状态互相污染**：策略 A 改了全局变量，策略 B 莫名其妙出错
- **一个崩溃全完蛋**：策略 A 网络超时抛异常，策略 B/C 还没跑就被迫退出
- **没法水平扩展**：顺序执行耗时 = A + B + C，无法利用多核

解决思路：**每个策略一个独立进程 + 统一配置 + 独立日志 + 结果回传**。

## 二、本文你将得到什么

1. **策略工厂模式**：新增策略只需注册，不改主流程代码
2. **multiprocessing 进程隔离**：每个策略独立 PID、独立内存、互不影响
3. **独立日志体系**：按策略分文件，排查时直奔主题
4. **统一配置注册表**：策略参数外置，调整无需改代码

## 三、策略工厂：新增策略只需一行注册

把策略实现和策略调度解耦：

```python
import pandas as pd

def strategy_sma(df, short=5, long=20):
    """SMA 金叉死叉策略"""
    df = df.copy()
    df["sma_short"] = df["c"].rolling(short).mean()
    df["sma_long"] = df["c"].rolling(long).mean()
    df["signal"] = 0
    df.loc[df["sma_short"] > df["sma_long"], "signal"] = 1
    df.loc[df["sma_short"] <= df["sma_long"], "signal"] = -1
    trades = df[df["signal"] != df["signal"].shift(1).fillna(0)]
    return {
        "name": "SMA_Cross",
        "last_signal": int(df["signal"].iloc[-1]),
        "trade_count": len(trades),
    }

def strategy_rsi(df, period=14, oversold=30):
    """RSI 超卖反弹策略"""
    df = df.copy()
    delta = df["c"].diff()
    gain = delta.where(delta > 0, 0.0)
    loss = (-delta).where(delta < 0, 0.0)
    rs = gain.rolling(period).mean() / loss.rolling(period).mean().replace(0, 1e-9)
    df["rsi"] = 100 - (100 / (1 + rs))
    triggers = df[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` 给每个策略分配独立进程，崩溃也不互相影响：

```python
import multiprocessing as mp
import logging

def run_strategy(name, code, params, log_path, result_queue):
    # 每个进程独立日志
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(name)s] %(message)s",
        handlers=[
            logging.FileHandler(log_path, encoding="utf-8"),
            logging.StreamHandler(),
        ],
    )
    log = logging.getLogger(name)
    log.info(f"启动 strategy={name} code={code}")

    # 拉取数据（内部实现省略）
    df = fetch_kline(code)
    func = STRATEGIES.get(name)
    result = func(df, **params)
    result["ok"] = True
    log.info(f"完成 result={result}")
    result_queue.put(result)
```

主调度入口：

```python
config = [
    {"name": "sma", "params": {"short": 5, "long": 20}, "log": "strategy_sma.log"},
    {"name": "rsi", "params": {"period": 14, "oversold": 30}, "log": "strategy_rsi.log"},
]

result_queue = mp.Queue()
processes = []
for cfg in config:
    p = mp.Process(target=run_strategy,
                   args=(cfg["name"], "600519", cfg["params"], cfg["log"], result_queue))
    p.start()
    processes.append(p)

# 等待全部完成
for p in processes:
    p.join(timeout=60)

# 收集结果
results = []
while not result_queue.empty():
    results.append(result_queue.get())
```

实测输出：

```
[start] sma -> pid=9476, log=strategy_sma.log
[start] rsi -> pid=3388, log=strategy_rsi.log

[ok] RSI_Bounce: last_rsi=50.79, trigger_count=0
[ok] SMA_Cross: last_signal=1, trade_count=1

strategy_sma.log: 488 bytes, 3 lines
strategy_rsi.log: 407 bytes, 3 lines
```

两个策略并行运行，PID 完全独立，日志分文件存储。

## 五、统一配置注册表：参数外置

把策略列表和参数写进 JSON 配置，主代码只负责读取和执行：

```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}
}
```

主代码加载：

```python
import json

with open("config.json", "r", encoding="utf-8") as f:
    cfg = json.load(f)

for strategy_cfg in cfg["strategies"]:
    # 直接按配置启动，无需硬编码
    print(f"注册策略: {strategy_cfg['name']} params={strategy_cfg['params']}")
```

**好处**：调参不用改 Python 代码，改 JSON 即可；Git 历史清晰，能追踪「哪次参数变更带来了收益变化」。

## 六、常见坑

1. **Queue 没消费完就退出**：主进程 `join()` 后，一定要把所有结果从 Queue 取出，否则子进程可能处于僵尸状态。
2. **日志Handler重复追加**：如果在循环里多次 `logging.basicConfig()`，同一个进程的日志会被重复打印。确保每个进程只配置一次。
3. **Windows 下 multiprocessing 必须加 `if __name__ == "__main__"`**：否则子进程会递归导入主模块，导致无限启动。
4. **共享 DataFrame 的陷阱**：不要试图把大 DataFrame 通过 Queue 传来传去，要么每个进程自己拉数据，要么用共享内存（`multiprocessing.shared_memory`，Python 3.8+）。

## 七、小结

本文把「多策略并行」从「一个脚本里顺序跑」升级为「工厂 + 进程隔离 + 配置驱动」的工程架构：

- **策略工厂**：新增策略只需注册，主流程零改动
- **进程隔离**：每个策略独立 PID、独立日志、独立崩溃域
- **配置外置**：参数放 JSON，版本可控、热更新友好
- **结果回传**：Queue 收集各策略结果，主进程统一汇总

下一篇（#06）我们将推进到「从本地脚本到云端部署」——给你一个量化系统的最小可运行架构模板，拿到就能用。

---

**免责声明**：本文仅供技术学习交流，不构成任何投资建议。量化策略回测表现不代表未来收益，投资有风险，决策需谨慎。

**代码与文档**：https://github.com/MaiRuiApi


一个策略跑得好，不代表三个策略一起跑也能好。日志互相覆盖、数据互相污染、一个策略崩溃把整组进程全带崩——本篇用 Python 的 multiprocessing + 策略工厂 + 独立日志，把「多策略并行」做成干净可扩展的工程模块。

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

编程语言

后端

开发工具

本文用Python的multiprocessing和策略工厂，解决多策略并行时的日志覆盖、数据污染与崩溃连锁问题。通过独立进程隔离、按策略分文件日志和JSON统一配置，实现干净可扩展的工程模块。同时给出策略注册、Queue结果回传和Windows启动等核心注意事项。

文件存储

代码解释

Windows

Python

JSON

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

【Python量化系统工程实战 #05】多策略并行怎么不打架？进程隔离 + 配置化策略工厂-腾讯云开发者社区-腾讯云

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

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

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐