我如何构建用于算法纸面交易的隔夜策略锦标赛系统

Source: Dev.to

请提供您希望翻译的正文内容（除代码块和 URL 之外），我将按照要求把它翻译成简体中文并保留原有的格式。

测试交易策略的问题

作为交易员，测试多个策略非常残酷。手动回测慢且不一致，而且你无法在睡觉时同时运行 10 个策略。大多数人最终只能凭直觉做决定，却伪装成分析。

我构建了 TradeSight 来解决这个问题：一个隔夜策略锦标赛，能够并行运行你的策略，依据真实的绩效指标对其排名，并在早晨提供排行榜。

核心思路很简单——让你的策略在真实的市场数据上相互竞争（通过 Alpaca 进行纸面交易），按夏普比率 + 胜率进行排名，让最好的策略存活下来。可以把它想象成进化压力，只不过对象是交易算法。

每晚

1. 所有已注册的策略为观察列表中的代码获取 OHLCV 数据。
2. 每个策略使用其自有的指标逻辑计算买入/卖出信号。
3. 纸面订单提交至 Alpaca。
4. 在早晨收盘时，对结果进行评分并在仪表盘上排名。

四个组件让它保持简洁

• Flask Dashboard – 配置锦标赛，查看实时仓位，排行榜。
• Strategy Runner – 并行执行策略（使用线程而非异步——对该场景更简洁）。
• Alpaca Integration – 用于订单提交和仓位跟踪的纸面交易 API。
• Cron Automation – 在收盘时启动锦标赛，开盘时进行评分。

项目结构

TradeSight/
├── app/
│   ├── dashboard.py      # Flask routes
│   ├── runner.py         # Strategy execution engine
│   └── alpaca_client.py  # Alpaca API wrapper
├── strategies/
│   ├── base.py           # Strategy base class
│   ├── macd_strategy.py
│   └── rsi_strategy.py
└── cron/
    └── overnight_run.sh

策略基类示例

# strategies/base.py
from abc import ABC, abstractmethod

class Strategy(ABC):
    name: str

    def __init__(self, alpaca):
        self.alpaca = alpaca

    @abstractmethod
    def compute_signals(self, ohlcv_data):
        """Return array of 1 (buy), -1 (sell), 0 (hold)."""
        pass

    @abstractmethod
    def submit_orders(self, signals, symbol, qty=10):
        """Submit paper orders based on signals."""
        pass

MACD交叉实现

# strategies/macd_strategy.py
import numpy as np
from strategies.base import Strategy

class MACDStrategy(Strategy):
    name = "MACD Crossover"

    def compute_signals(self, ohlcv_data):
        close = ohlcv_data['close'].values

        # EMA calculations
        ema12 = self._ema(close, 12)
        ema26 = self._ema(close, 26)
        macd_line = ema12 - ema26
        signal_line = self._ema(macd_line, 9)

        # Crossover detection
        signals = np.zeros(len(close))
        for i in range(1, len(macd_line)):
            if macd_line[i] > signal_line[i] and macd_line[i-1] = signal_line[i-1]:
                signals[i] = -1  # bearish crossover
        return signals

    def submit_orders(self, signals, symbol, qty=10):
        latest = signals[-1]
        if latest == 1:
            self.alpaca.submit_order(symbol=symbol, qty=qty, side='buy', type='market')
        elif latest == -1:
            self.alpaca.submit_order(symbol=symbol, qty=qty, side='sell', type='market')

内置策略

TradeSight 附带六个即用型策略：

1. MACD 交叉
2. RSI 超买/超卖
3. 布林带压缩
4. EMA 带
5. 成交量激增 + 价格确认
6. 均值回归（基于 Z‑score）

运行锦标赛

import pandas as pd
import numpy as np
from alpaca_trade_api import TimeFrame

def run_tournament(symbols, strategies, alpaca):
    results = []

    for strategy in strategies:
        strategy_results = []
        for symbol in symbols:
            # Fetch OHLCV from Alpaca
            bars = alpaca.get_bars(symbol, TimeFrame.Minute, limit=200)
            ohlcv = pd.DataFrame([b._raw for b in bars])

            # Get signals
            signals = strategy.compute_signals(ohlcv)

            # Submit paper orders
            strategy.submit_orders(signals, symbol)

            # Track for scoring
            strategy_results.append({
                'symbol': symbol,
                'signal': signals[-1],
                'entry_price': ohlcv['close'].iloc[-1]
            })

        results.append({'strategy': strategy.name, 'trades': strategy_results})

    return results

评分策略

def score_strategy(trades):
    winning_trades = [t for t in trades if t['pnl'] > 0]
    win_rate = len(winning_trades) / len(trades) if trades else 0

    returns = [t['pnl_pct'] for t in trades]
    sharpe = (np.mean(returns) / np.std(returns)) * np.sqrt(252) if returns else 0

    # Combined score: weights Sharpe more than win rate
    score = (sharpe * 0.6) + (win_rate * 0.4)
    return {'sharpe': sharpe, 'win_rate': win_rate, 'score': score}

Flask 仪表板显示实时排行榜、当前未平仓头寸以及历史锦标赛结果，让您能够随时间观察策略的改进（或失败）。

添加自定义策略

1. 将文件放入 strategies/ 目录。
2. 继承自 Strategy。
3. 实现 compute_signals 和 submit_orders。

class MyCustomStrategy(Strategy):
    name = "My Custom Strategy"

    def compute_signals(self, ohlcv_data):
        # Your logic here
        pass

    def submit_orders(self, signals, symbol, qty=10):
        # Your execution logic here
        pass

比赛运行器会在启动时自动发现所有策略类。

通过夜间锦标赛获得的洞见

运行几周的夜间锦标赛让我对策略行为的理解超过了数月的手动回测。

• 执行时机很重要 – 在纸面上看起来不错的策略往往在实时数据上崩溃，因为订单延迟和滑点。纸上交易并不完美，但比使用完美事后视角的回测更接近真实。
• 一致性胜过峰值表现 – 一个在 60 % 锦标赛中获胜的策略，比一次惊艳获胜而其余全部失败的策略更有价值。

源代码

完整源码可在 github.com/rmbell09-lang/tradesight 获取。欢迎点星和提交 PR，尤其是新的策略实现。