
跑一個量化策略需要的不只是策略本身——還需要一套支撐它執行、監控它狀態、在出問題時及時告警的工具鏈。這篇文章分享我的完整工具棧,從策略執行到告警通知。
整體架構
策略層(Python)
↓ 交易指令
Binance API
↓ 執行結果
狀態檔案 / CF KV
↓ 指標
Prometheus + node_exporter
↓ 儲存
Grafana 視覺化
↓ 告警規則
Alertmanager → Telegram
策略執行層
技術棧
- 語言:Python 3.11
- Binance SDK:
python-binance - 定時任務:APScheduler 或系統 cron
- 依賴管理:
requirements.txt+ venv
目錄結構
quant/
├── strategy/
│ ├── main.py # 主策略邏輯(訊號生成 + 下單)
│ ├── signals.py # 指標計算(技術指標庫)
│ └── risk.py # 倉位管理、止損邏輯
├── exchange/
│ └── binance_client.py # Binance API 封裝
├── monitor/
│ └── exporter.py # 狀態指標寫入 Prometheus 格式
├── config/
│ └── settings.py # 配置(從環境變數讀取)
└── run.py # 入口
關鍵程式碼:狀態指標暴露
策略執行時,把關鍵指標寫成 Prometheus textfile 格式,供 node_exporter 抓取:
# monitor/exporter.py
import os
import time
METRICS_FILE = "/var/lib/node_exporter/textfile_collector/quant.prom"
def write_metrics(data: dict):
"""
data 示例:
{
"pnl_total": 12.5, # 累計收益率 %
"pnl_today": 0.8, # 今日收益率 %
"position": 1, # 1=多, -1=空, 0=空倉
"last_signal_ts": 1776200000, # 最後訊號時間戳
"last_run_ts": 1776235000, # 策略最後執行時間
}
"""
lines = [
'# HELP quant_pnl_total 累計收益率(百分比)',
'# TYPE quant_pnl_total gauge',
f'quant_pnl_total {data["pnl_total"]}',
'',
'# HELP quant_pnl_today 今日收益率(百分比)',
'# TYPE quant_pnl_today gauge',
f'quant_pnl_today {data["pnl_today"]}',
'',
'# HELP quant_position 當前持倉方向 (1=多,-1=空,0=空倉)',
'# TYPE quant_position gauge',
f'quant_position {data["position"]}',
'',
'# HELP quant_last_signal_timestamp 最後訊號時間',
'# TYPE quant_last_signal_timestamp gauge',
f'quant_last_signal_timestamp {data["last_signal_ts"]}',
'',
'# HELP quant_last_run_timestamp 策略最後執行時間',
'# TYPE quant_last_run_timestamp gauge',
f'quant_last_run_timestamp {int(time.time())}',
'',
]
tmp = METRICS_FILE + ".tmp"
with open(tmp, "w") as f:
f.write("\n".join(lines))
os.rename(tmp, METRICS_FILE)
伺服器配置
選擇
量化策略對伺服器的核心要求:
- 穩定:不能隨意宕機,24/7 執行
- 延遲低:離 Binance 伺服器近(新加坡或東京),下單延遲更低
- 獨立:不和其他高負載任務共享
我用 AWS 新加坡節點(aws-sg),Binance 主伺服器在新加坡,下單延遲 1-5ms。
systemd 守護程序
不要用 nohup 跑策略,用 systemd:
# /etc/systemd/system/quant-strategy.service
[Unit]
Description=Quant Strategy
After=network.target
[Service]
Type=simple
User=ubuntu
WorkingDirectory=/home/ubuntu/quant
ExecStart=/home/ubuntu/quant/venv/bin/python run.py
Restart=always
RestartSec=10
EnvironmentFile=/home/ubuntu/quant/.env
[Install]
WantedBy=multi-user.target
systemctl enable quant-strategy
systemctl start quant-strategy
systemctl status quant-strategy
策略崩潰會自動重啟,重啟間隔 10 秒。
環境變數管理
API Key 放 .env 檔案,不提交到 Git:
# .env(只有 600 許可權)
BINANCE_API_KEY=xxxx
BINANCE_API_SECRET=xxxx
TELEGRAM_BOT_TOKEN=xxxx
TELEGRAM_CHAT_ID=xxxx
# settings.py
from dotenv import load_dotenv
import os
load_dotenv()
BINANCE_API_KEY = os.environ["BINANCE_API_KEY"]
BINANCE_API_SECRET = os.environ["BINANCE_API_SECRET"]
Prometheus 監控配置
在 prometheus.yml 新增策略伺服器的 node_exporter:
scrape_configs:
- job_name: node-remote-linux
static_configs:
- targets:
- 你的策略伺服器IP:9100
node_exporter 預設會自動採集 /var/lib/node_exporter/textfile_collector/*.prom 裡的指標,不需要額外配置。
Grafana Dashboard
量化策略的核心監控面板包含:
收益區(上方)
- 累計收益率時序曲線
- 今日收益 Stat 卡片
- 當前持倉方向(多/空/空倉)
風險區(中間)
- 最大回撤(Max Drawdown)
- 近 30 日日收益柱狀圖
- 夏普比率(年化)
健康區(下方)
- 策略最後執行時間(距現在多久)
- 最後訊號時間(距現在多久)
- 心跳狀態(綠色=正常,紅色=異常)
關鍵面板:策略心跳
# Grafana PromQL
(time() - quant_last_run_timestamp) / 60
這個查詢顯示策略最後執行距現在多少分鐘。配合閾值顏色:
- < 65 分鐘:綠色(正常,小時級策略)
- 65-120 分鐘:黃色(輕微延遲)
120 分鐘:紅色(策略可能掛了)
告警配置
Alertmanager + Telegram
在 alerts.yml 裡定義告警規則:
groups:
- name: quant-alerts
rules:
# 策略心跳超時
- alert: QuantHeartbeatTimeout
expr: (time() - quant_last_run_timestamp) > 7200
for: 5m
labels:
severity: critical
annotations:
summary: "量化策略心跳超時"
description: "策略超過 2 小時未執行,請檢查伺服器狀態"
# 今日大幅虧損
- alert: QuantLargeLoss
expr: quant_pnl_today < -3
for: 1m
labels:
severity: warning
annotations:
summary: "今日虧損超過 3%"
description: "當日收益率 {{ $value }}%,請關注"
Telegram 告警模板(在 alertmanager.yml 裡配置):
receivers:
- name: telegram
telegram_configs:
- bot_token: "你的BOT_TOKEN"
chat_id: 你的CHAT_ID
message: |
🚨 {{ .GroupLabels.alertname }}
{{ range .Alerts }}
{{ .Annotations.summary }}
{{ .Annotations.description }}
{{ end }}
在策略程式碼裡直接推送告警
除了 Alertmanager,策略程式碼裡也可以直接推 Telegram:
import requests
def send_telegram(message: str):
bot_token = os.environ["TELEGRAM_BOT_TOKEN"]
chat_id = os.environ["TELEGRAM_CHAT_ID"]
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
requests.post(url, json={"chat_id": chat_id, "text": message})
# 下單後推送
send_telegram(f"📊 訊號:{direction}\n價格:{price}\n倉位:{size}")
總結:工具選型邏輯
| 工具 | 替代方案 | 選這個的原因 |
|---|---|---|
| Python | Go、JavaScript | 量化生態最豐富(pandas、ta-lib 等) |
| Binance python-binance | 自己封裝 REST | 官方 SDK,省去重複造輪子 |
| systemd | supervisor、PM2 | 系統自帶,穩定,日誌集中管理 |
| Prometheus + node_exporter | InfluxDB + Telegraf | 已有監控棧,複用,不引入新元件 |
| Alertmanager | 自己寫告警指令碼 | 支援規則、靜默、分組,功能完整 |
| Telegram | 郵件、釘釘 | 手機推送及時,API 簡單 |
→ 檢視 北極星策略介紹 → 檢視 Binance API 從零到第一筆下單 → 檢視 Prometheus + Grafana VPS 監控搭建
📷 封面圖 · Flux Schnell AI 生成 · panshi.io 品牌風格