交易系统架构对比分析报告

对比项目：passivbot（1,860 stars / 625 forks）
分析日期：2026-02-08
版本：v7.8.1

一、参考项目概览

Passivbot 是一个成熟的多交易所永续合约自动交易系统，采用 Python（编排/IO）+ Rust（计算引擎） 混合架构。核心策略为逆势网格做市（Martingale 变种），不预测价格走势，通过网格加仓拉低均价 + 反弹获利平仓。支持 Hyperliquid、Binance、Bybit、OKX、Bitget、GateIO、Kucoin 等 7+ 交易所。

项目结构

src/
├── main.py                        # 入口（Python 版本检查 + Rust 编译检查）
├── passivbot.py                   # 核心 Bot 类（~2500 行）
├── candlestick_manager.py         # K线数据管理（磁盘分片/原子写入/gap处理）
├── fill_events_manager.py         # 成交事件管理（跨进程限流/gap置信度）
├── config_utils.py                # 配置加载/转换/覆盖白名单
├── config_transform.py            # 配置格式转换
├── backtest.py                    # 回测引擎（调用 Rust）
├── backtest_suite.py              # 多场景回测套件
├── optimize.py                    # DEAP 进化算法优化器
├── optimize_suite.py              # 多场景优化
├── pareto_core.py                 # 帕累托前沿分析
├── pareto_store.py                # 帕累托结果存储
├── downloader.py                  # 历史数据下载
├── pure_funcs.py                  # 纯函数工具（无副作用）
├── utils.py                       # 通用工具
├── rust_utils.py                  # Rust FFI 编译/加载/同步
├── logging_setup.py               # 4级日志（WARN/INFO/DEBUG/TRACE）
├── exchanges/
│   ├── ccxt_bot.py                # CCXT 基类（Hook 分类法）
│   ├── hyperliquid.py             # Hyperliquid 适配器（514 行）
│   ├── binance.py                 # Binance 适配器
│   ├── bybit.py                   # Bybit 适配器
│   ├── bitget.py                  # Bitget 适配器
│   ├── okx.py                     # OKX 适配器
│   ├── gateio.py                  # GateIO 适配器
│   ├── kucoin.py                  # Kucoin 适配器
│   ├── paradex.py                 # Paradex 适配器
│   └── defx.py                    # Defx 适配器
├── optimization/
│   ├── bounds.py                  # 参数边界定义
│   ├── config_adapter.py          # 优化器配置适配
│   └── deap_adapters.py           # DEAP 适配器
└── tools/
    ├── fill_events_dash.py        # 成交事件仪表盘
    ├── pareto_dash.py             # 帕累托可视化
    ├── iterative_backtester.py    # 迭代回测工具
    └── ...                        # 其他辅助工具

passivbot-rust/src/
├── orchestrator.rs                # 订单规划编排器（核心）
├── entries.rs                     # 入场订单计算
├── closes.rs                      # 出场订单计算
├── trailing.rs                    # 追踪订单逻辑
├── risk.rs                        # 风控引擎（TWEL/WEL/unstuck）
├── backtest.rs                    # Rust 回测引擎
├── coin_selection.rs              # Forager 动态选币
├── analysis.rs                    # 回测分析（50+ 指标）
├── types.rs                       # 核心数据类型
├── utils.rs                       # 数学工具函数
├── constants.rs                   # 常量定义
├── python.rs                      # PyO3 绑定层
└── lib.rs                         # 库入口

configs/
├── template.json                  # 完整配置模板（~350 行）
├── approved_coins.json            # 允许交易的币种列表
├── custom_endpoints.json.example  # 自定义 API 端点
└── examples/                      # 示例配置

tests/                             # 80+ 测试文件

核心数据流

JSON Config → config_utils.load_config() (validated)
    ↓
Passivbot.__init__()
    ↓
┌─────────────────────────────────────────────────┐
│       EXECUTION LOOP (30s cycle)                │
│                                                 │
│  update_pos_oos_pnls_ohlcvs()                   │
│       ↓ (交易所→本地状态同步)                      │
│  CandlestickManager.get_candles()               │
│       ↓ (K线数据 + EMA 计算)                      │
│  pbr.calc_orders_py()  [Rust Orchestrator]      │
│       ↓ (理想订单集)                              │
│  filter_orders()  (actual vs ideal diff)        │
│       ↓                                         │
│  execute_cancellations() → execute_creations()  │
│       ↓                                         │
│  health_summary() (每15分钟)                     │
└─────────────────────────────────────────────────┘

技术栈

二、值得借鉴的 10 大设计模式

1. Unstuck 机制（被套仓位渐进解套）⭐⭐⭐⭐⭐

参考项目做法：

// risk.rs — calc_unstucking_action()
// 当某个仓位的 wallet_exposure 超过 unstuck_threshold * WEL 时触发

fn calc_unstucking_action(position, balance, config) -> Option<CloseOrder> {
    // 1. 计算亏损容忍预算 = (当前余额 - 峰值余额 * (1 - loss_allowance_pct))
    // 2. 如果预算 > 0，在 EMA band 价格处生成小额止损单
    // 3. 优先处理离市场价最近的被套仓位（unstuck 成本最低）
    // 4. close_pct 控制每次解套比例，防止一次性大额止损
}

核心逻辑：

• 峰值余额追踪：记录历史最高余额，亏损预算 = 当前余额 - 峰值 × (1 - loss_allowance_pct)
• 优先级排序：多个被套仓位时，优先处理 |entry_price - market_price| 最小的（止损成本最低）
• 渐进式：每次只平一小部分（close_pct），不一次性割肉
• EMA 锚定：在 EMA band 处挂单而非市价止损，争取更好价格

当前项目现状：

• risk_manager.py 有 max_drawdown_pct 和 daily_loss_limit 检查
• 触发后是粗暴的全量平仓（通过 CircuitBreaker 或 KillSwitch）
• 没有"渐进式小额止损"的概念 — 要么持有要么全平

改进方向：

• 对配对交易特别有价值 — 当一腿被深度套牢时，可渐进减仓而非等待均值回归
• 结合已有 daily_loss_limit 作为总预算，在预算内自动渐进止损
• 按 |entry_price - mark_price| / entry_price 排序决定哪个仓位优先解套

2. 双层敞口限制系统（WEL + TWEL）⭐⭐⭐⭐⭐

参考项目做法：

// risk.rs — gate_entries_by_twel()
// WEL = Wallet Exposure Limit（单仓位敞口上限）
// TWEL = Total Wallet Exposure Limit（全账户总敞口上限）

fn gate_entries_by_twel(orders, positions, balance, twel_threshold) -> Vec<Order> {
    let current_total_exposure = sum(positions.map(|p| p.notional / balance));

    if current_total_exposure >= twel_threshold {
        // 移除所有新入场单
        return orders.filter(|o| !o.is_entry());
    }

    // 二分查找：在 TWEL 限额内，最大可填充数量
    let max_fillable = binary_search_max_qty(remaining_budget, price, qty_step);
    // 优先移除离市场价最远的入场单
    remove_furthest_from_market_first(orders, max_fillable);
}

TWEL Enforcer（自动减仓）：

fn calc_twel_enforcer_actions(positions, balance, config) -> Vec<CloseOrder> {
    // 当 total_exposure > twel_enforcer_threshold 时自动减仓
    // 优先减 price_diff 最小的仓位（最不亏）
    // 每个仓位有 floor exposure 保护，不会减到 0
    // 级联逻辑：如果最优先仓位已到 floor，转向下一个
}

当前项目现状：

• risk_manager.py 有 max_position_notional（单仓位上限）和 max_open_positions（持仓数上限）
• 没有"全账户总敞口"的概念 — 理论上可以同时持有多对配对交易，总敞口超出安全范围
• 没有自动减仓机制，超限只能拒绝新单

改进方向：

• 引入 TWEL 概念，确保所有配对交易的总敞口不超过账户权益的 N%
• 当总敞口超标时，TWEL Enforcer 自动减仓最不亏的仓位
• 配对交易场景下 WEL 可用于控制单对仓位的最大敞口

3. 订单容差匹配（避免无效撤单/挂单）⭐⭐⭐⭐⭐

参考项目做法：

# pure_funcs.py — filter_orders()
def filter_orders(actual_orders, ideal_orders, price_tol=0.001, qty_tol=0.01):
    """比较实际挂单 vs 理想挂单，只操作差异部分"""
    to_cancel = []
    to_create = []

    for actual in actual_orders:
        matched = find_matching_ideal(actual, ideal_orders, price_tol, qty_tol)
        if not matched:
            to_cancel.append(actual)

    for ideal in ideal_orders:
        matched = find_matching_actual(ideal, actual_orders, price_tol, qty_tol)
        if not matched:
            to_create.append(ideal)

    return to_cancel, to_create

# passivbot.py — orders_matching()
def orders_matching(order_a, order_b, price_tol=0.003, qty_tol=0.02):
    """两个订单在容差范围内视为相同"""
    if order_a["side"] != order_b["side"]:
        return False
    if abs(order_a["price"] - order_b["price"]) / order_a["price"] > price_tol:
        return False
    if abs(order_a["qty"] - order_b["qty"]) / max(order_a["qty"], order_b["qty"]) > qty_tol:
        return False
    return True

当前项目现状：

• executor.py 每次执行都是「取消全部 → 重新挂单」的模式
• 即使价格/数量只变了 0.01%，也会触发撤单+重挂
• 高频操作浪费 API 调用额度，增加被限流风险

改进方向：

• 实现 ideal_vs_actual 差异对比，只撤/挂有变化的单
• 设置合理容差（如价格 ±0.3%，数量 ±2%）
• 对 PURR 这样低流动性品种尤其重要 — 减少不必要的订单变动对薄盘口的冲击

4. TradingMode 状态机（5 种运行模式）⭐⭐⭐⭐

参考项目做法：

// orchestrator.rs
enum TradingMode {
    Normal,         // 正常交易（入场 + 出场）
    TpOnly,         // 只允许止盈平仓，不新开仓
    GracefulStop,   // 优雅停止：不开新仓，有序平现有仓
    Panic,          // 紧急模式：市价全平
    Manual,         // 人工接管：bot 不操作
}

// 在 Orchestrator 中根据 mode 过滤订单
fn compute_orders(input: OrchestratorInput) -> Vec<IdealOrder> {
    match input.trading_mode {
        Normal => compute_entries_and_closes(input),
        TpOnly => compute_closes_only(input),
        GracefulStop => compute_graceful_close(input),
        Panic => compute_panic_close_all(input),
        Manual => vec![],  // 不生成任何订单
    }
}

当前项目现状：

• KillSwitch 只有 on/off 两态
• 没有"只平仓不开仓"的中间态
• 没有"优雅停止"的概念 — 触发 KillSwitch 就是立即全停

改进方向：

• 引入 TradingMode 枚举替代简单的 kill_switch 布尔值
• TpOnly：日损限额接近阈值时自动切换，只等待现有仓位止盈
• GracefulStop：维护/升级前手动触发，不开新仓但允许有序平仓
• Panic：等同于当前 KillSwitch，保留作为最后手段
• Manual：远程调试时暂停 bot，手动操作

5. 进化算法参数优化器 ⭐⭐⭐⭐

参考项目做法：

# optimize.py — 使用 DEAP 框架（NSGA-II 多目标优化）
class ConstraintAwareFitness:
    """带约束的适应度评估"""
    # 约束条件：最大回撤 < 30%, 亏损/利润比 < 0.5, 恢复时间 < 72h
    # 评估指标：ADG（日均增长）、Sharpe、Sortino、Omega...

# 优化流程：
# 1. 定义参数搜索空间（bounds.py）
# 2. 随机初始化种群（250 个个体）
# 3. 迭代 500K 次：
#    a. 交叉 + 变异生成新个体
#    b. 回测评估（调用 Rust 引擎，毫秒级）
#    c. Pareto 非支配排序 + 拥挤度距离
#    d. 淘汰劣势个体
# 4. 输出 Pareto 前沿上的最优配置集

// configs/template.json — 优化参数示例
"optimize": {
    "population_size": 250,
    "iters": 500000,
    "mutation_probability": 0.2,
    "crossover_probability": 0.7,
    "scoring": ["mdg", "sharpe_ratio"],
    "limits": {
        "lower_bound_drawdown_worst": 0.30,
        "lower_bound_loss_profit_ratio": 0.50,
        "upper_bound_time_at_max_exposure": 0.50,
        "upper_bound_peak_recovery_hours": 72.0
    }
}

当前项目现状：

• 配对交易参数（Z-score 阈值、EMA 窗口、止损百分比等）全靠手动调参
• 没有系统化的回测框架
• 参数选择基于经验而非数据验证

改进方向：

• 构建配对交易回测引擎（利用 TimescaleDB 中的历史数据）
• 可优化参数：Z-score 开仓/平仓阈值、EMA 窗口组合、止损/止盈比例、仓位大小
• 评估指标：Sharpe ratio、最大回撤、胜率、盈亏比
• 约束条件：回撤 < X%、单日亏损 < Y、恢复时间 < Z 小时
• 不必用 Rust，Python + NumPy 回测对配对交易足够快

6. 原子文件操作 + 跨进程锁 ⭐⭐⭐⭐

参考项目做法：

# candlestick_manager.py — 原子写入
def _atomic_write_bytes(path, data):
    """防止写入中断导致文件损坏"""
    tmp_path = path.with_suffix(".tmp")
    with open(tmp_path, "wb") as f:
        f.write(data)
        f.flush()
        os.fsync(f.fileno())  # 确保写入磁盘
    os.replace(tmp_path, path)  # 原子替换

# 跨进程缓存锁
import portalocker
class CacheFileLock:
    """可重入的分布式文件锁"""
    # 自动清理过期锁（防止进程崩溃遗留死锁）
    # 300 秒超时
    # 支持多 bot 实例共享 OHLCV 缓存

当前项目现状：

• 主要通过 TimescaleDB 持久化，不太涉及文件写入
• 但 .env 配置文件、日志文件等无原子写入保护
• 单实例运行，不考虑多进程竞争

改进方向：

• 如果未来支持多策略实例共享数据，需要引入跨进程锁
• 对于任何文件写入（配置热更新、状态快照等），使用原子写入模式
• 当前优先级较低，但架构意识值得保留

7. Warmup 交错预热系统 ⭐⭐⭐⭐

参考项目做法：

# passivbot.py — warmup_candles_staggered()
async def warmup_candles_staggered(self, symbols, concurrency=4):
    """交错式 K 线预热，防止 API 风暴"""
    semaphore = asyncio.Semaphore(concurrency)

    async def fetch_one(symbol):
        async with semaphore:
            jitter = random.uniform(0.1, 0.5)  # 随机抖动
            await asyncio.sleep(jitter)

            # 每个 symbol 计算需要的 warmup 窗口
            window = compute_warmup_window(symbol, ema_spans)
            await self.cm.fetch_candles(symbol, window)

    # 并发但受控地预热所有 symbol
    await asyncio.gather(*[fetch_one(s) for s in symbols])

当前项目现状：

• realtime_kline_service_base.py 启动时批量获取 K 线
• 没有并发控制和抖动机制
• 如果同时请求多个品种的多个周期，可能触发 API 限流

改进方向：

• 用 asyncio.Semaphore 控制 K 线请求并发数
• 添加随机抖动（jitter）防止请求聚集
• 根据 EMA 窗口大小动态计算每个品种需要的预热数据量

8. 定期健康摘要 + 内存监控 ⭐⭐⭐⭐

参考项目做法：

# passivbot.py — _maybe_log_health_summary()
def _maybe_log_health_summary(self):
    """每 15 分钟输出一次系统健康报告"""
    if time.time() - self._last_health_summary < 900:
        return

    summary = {
        "uptime": format_duration(time.time() - self._start_time),
        "orders_placed": self._health_orders_placed,
        "fills": self._health_fills,
        "errors": self._health_errors,
        "ws_reconnects": self._health_ws_reconnects,
        "rate_limit_hits": self._health_rate_limits,
        "positions": len(self.positions),
        "unrealized_pnl": sum(p.upnl for p in self.positions.values()),
    }

    # 内存快照
    import resource
    rss_mb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024 / 1024
    summary["memory_rss_mb"] = round(rss_mb, 1)
    summary["cm_cache_entries"] = self.cm.cache_size()
    summary["async_tasks"] = len(asyncio.all_tasks())

    logging.info(f"[health] {json.dumps(summary)}")
    self._reset_health_counters()

当前项目现状：

• 通过飞书通知发送关键事件（开仓/平仓/止损）
• 没有定期的系统级健康报告
• 没有内存使用监控 — 长时间运行后内存泄漏难以发现

改进方向：

• 每 N 分钟生成健康摘要（运行时长、下单数、成交数、错误数、WS 重连数、内存使用）
• 通过飞书定期推送（而非只在异常时通知）
• 追踪 WS 管理器的连接健康状态
• 设置内存使用告警阈值，及早发现泄漏

9. 交易所适配器 Hook 分类法 ⭐⭐⭐

参考项目做法：

# exchanges/ccxt_bot.py — 基于 Hook 的多交易所适配
class CCXTBot:
    """通用 CCXT 基类，子类通过覆盖 Hook 适配不同交易所"""

    # can_* hooks — 能力查询
    def can_hedge_mode(self) -> bool: ...      # 是否支持双向持仓
    def can_watch_orders(self) -> bool: ...     # 是否支持 WS 订单推送

    # _do_* hooks — API 调用（子类覆盖实际接口差异）
    async def _do_fetch_balance(self) -> dict: ...
    async def _do_fetch_positions(self) -> list: ...
    async def _do_create_order(self, params) -> dict: ...

    # _get_* hooks — 数据提取（从原始响应中提取标准字段）
    def _get_position_size(self, raw) -> float: ...
    def _get_order_status(self, raw) -> str: ...

    # _normalize_* hooks — 数据转换（统一格式）
    def _normalize_position(self, raw) -> Position: ...
    def _normalize_order(self, raw) -> Order: ...

    # _build_* hooks — 配置构建
    def _build_ccxt_config(self) -> dict: ...
    def _build_order_params(self, order) -> dict: ...

# exchanges/hyperliquid.py — Hyperliquid 特化
class HyperliquidBot(CCXTBot):
    def can_hedge_mode(self) -> bool:
        return False  # Hyperliquid 只支持单向持仓

    def _do_create_order(self, params):
        # HIP-3 stock perps 特殊处理（TSLA, NVDA 等）
        if params["symbol"].startswith("xyz:"):
            # 特殊精度处理...

        # $10 minimum 错误自动恢复
        try:
            return await super()._do_create_order(params)
        except MinCostError:
            self.adjust_min_cost_on_error(params["symbol"])
            return await super()._do_create_order(params)

当前项目现状：

• HyperliquidExecutor 直接耦合 Hyperliquid SDK
• 没有交易所抽象层
• 如果未来需要接入 dYdX、GMX 等 DEX，需要大量重写

改进方向：

• Hook 分类法（can_* / _do_* / _get_* / _normalize_*）是比简单 ABC 接口更灵活的适配方案
• 子类只需覆盖差异部分，公共逻辑留在基类
• 可作为 P2 长期目标，当前单交易所无紧迫需求

10. 回测场景套件（Scenario-Based Backtesting）⭐⭐⭐

参考项目做法：

// configs/template.json — 多场景回测
"backtest": {
    "start_date": "2021-05-01",
    "end_date": "now",
    "starting_balance": 10000.0,
    "scenarios": {
        "base": {},
        "trump_troupe": {
            "coins": ["DOGE", "PEPE", "SHIB", "FLOKI"],
            "overrides": {"bot": {"long": {"n_positions": 4}}}
        },
        "pure_trailing": {
            "overrides": {"bot": {"long": {"entry_grid_spacing_pct": 100.0}}}
        },
        "twel_stress": {
            "overrides": {"bot": {"long": {"total_wallet_exposure_limit": 0.5}}}
        }
    }
}

回测分析输出（50+ 指标）：

// analysis.rs
struct Analysis {
    adg: f64,           // 日均增长率
    mdg: f64,           // 中位日增长
    sharpe: f64,        // 夏普比率
    sortino: f64,       // 索提诺比率
    omega: f64,         // 欧米茄比率
    calmar: f64,        // 卡尔玛比率
    sterling: f64,      // 斯特林比率
    drawdown_worst: f64, // 最大回撤
    drawdown_mean: f64,  // 平均回撤
    peak_recovery_hours: f64, // 峰值恢复时间
    n_fills: usize,     // 成交次数
    n_entries: usize,   // 入场次数
    n_closes: usize,    // 平仓次数
    time_at_max_exposure: f64, // 最大敞口时间占比
    loss_profit_ratio: f64,    // 亏损/利润比
    // ... 50+ 指标
}

当前项目现状：

• 没有系统化的回测框架
• 参数调优依靠实盘观察 + 手动调整
• 没有多场景压力测试（如极端行情、低波动期等）

改进方向：

• 利用 TimescaleDB 中的历史数据构建回测引擎
• 定义配对交易专用场景：高波动期、低波动期、趋势行情、震荡行情
• 关键指标：Sharpe ratio、最大回撤、胜率、持仓时间分布、Pair spread 回归时间
• 进化算法优化可作为 P2 目标，先实现基础回测

三、当前项目的独有优势

以下是当前项目已有但 passivbot 缺失的能力，不需要改动：

四、改进优先级建议

P0 — 高价值、可立即实施

P1 — 高价值、中等工作量

P2 — 中价值、可延后

五、关键设计理念对比

六、总结

Passivbot 是目前分析的 5 个项目中最成熟的（1,860 stars、6 年迭代、Rust 性能优化、80+ 测试文件）。其核心价值不在策略（Martingale 网格做市与配对交易完全不同），而在于工程化的交易运营体系。

最值得优先借鉴的 5 个点：

1. 订单容差匹配（3-4h）— 投入最小、收益最大。避免每个周期全量撤单+重挂，减少 API 调用，降低对薄盘口的冲击
2. TradingMode 状态机（4-6h）— 从"运行/停止"二态进化为 5 态，支持 TpOnly（接近日损限额时）和 GracefulStop（维护时）
3. Unstuck 渐进解套（6-8h）— 对配对交易被套仓位，不再只能全平或死扛，而是在亏损预算内渐进止损
4. 定期健康摘要（2-3h）— 每 15 分钟发送系统健康报告到飞书，含内存/错误/WS 状态
5. 基础回测框架（2-3d）— 利用 TimescaleDB 历史数据，验证参数有效性，从经验驱动升级为数据驱动