OKX 量化交易：五策略框架 + ATR 动态止损 + PySide6 仪表盘

pip install ccxt pandas numpy PySide6 pyqtgraph

quant/
├── main.py          # 主入口，启动 GUI + 策略引擎
├── engine.py        # 交易引擎（下单/平仓/持仓管理）
├── strategies/
│   ├── base.py      # 策略基类
│   ├── ema_cross.py
│   ├── boll.py
│   ├── macd.py
│   ├── rsi_rev.py
│   └── combo.py     # 组合投票策略
├── indicators.py    # 技术指标计算
└── dashboard.py     # PySide6 仪表盘

from abc import ABC, abstractmethod
import pandas as pd

class BaseStrategy(ABC):
    name: str = "base"
    weight: float = 1.0  # 组合策略权重

    def __init__(self, symbol: str, params: dict = None):
        self.symbol = symbol
        self.params = params or {}

    @abstractmethod
    def signal(self, df: pd.DataFrame) -> int:
        # 返回信号：1=做多, -1=做空, 0=观望
        ...

class EMACrossStrategy(BaseStrategy):
    name = "ema_cross"

    def __init__(self, symbol, params=None):
        super().__init__(symbol, params)
        self.fast = self.params.get("fast", 9)
        self.slow = self.params.get("slow", 21)

    def signal(self, df: pd.DataFrame) -> int:
        df["ema_fast"] = df["close"].ewm(span=self.fast).mean()
        df["ema_slow"] = df["close"].ewm(span=self.slow).mean()
        last = df.iloc[-1]
        prev = df.iloc[-2]
        if prev["ema_fast"] < prev["ema_slow"] and last["ema_fast"] > last["ema_slow"]:
            return 1   # 金叉做多
        if prev["ema_fast"] > prev["ema_slow"] and last["ema_fast"] < last["ema_slow"]:
            return -1  # 死叉做空
        return 0

import numpy as np

def atr(df: pd.DataFrame, period: int = 14) -> pd.Series:
    high, low, close = df["high"], df["low"], df["close"]
    tr = pd.concat([
        high - low,
        (high - close.shift()).abs(),
        (low  - close.shift()).abs()
    ], axis=1).max(axis=1)
    return tr.ewm(span=period).mean()

def calc_stop_loss(entry: float, side: str, df: pd.DataFrame,
                   atr_mult: float = 2.0) -> float:
    # 根据 ATR 动态计算止损价
    atr_val = atr(df).iloc[-1]
    if side == "long":
        return entry - atr_val * atr_mult
    else:
        return entry + atr_val * atr_mult

class BollStrategy(BaseStrategy):
    name = "boll"

    def signal(self, df: pd.DataFrame) -> int:
        period = self.params.get("period", 20)
        std_mult = self.params.get("std", 2.0)
        mid = df["close"].rolling(period).mean()
        std = df["close"].rolling(period).std()
        upper = mid + std_mult * std
        lower = mid - std_mult * std

        last_close = df["close"].iloc[-1]
        last_lower = lower.iloc[-1]
        last_upper = upper.iloc[-1]

        if last_close <= last_lower:
            return 1   # 触下轨，做多
        if last_close >= last_upper:
            return -1  # 触上轨，做空
        return 0

class ComboStrategy(BaseStrategy):
    name = "combo"

    def __init__(self, symbol, strategies: list):
        super().__init__(symbol)
        self.strategies = strategies  # [BaseStrategy, ...]

    def signal(self, df: pd.DataFrame) -> int:
        votes = sum(s.signal(df) * s.weight for s in self.strategies)
        total_weight = sum(s.weight for s in self.strategies)
        ratio = votes / total_weight
        if ratio >= 0.5:
            return 1
        if ratio <= -0.5:
            return -1
        return 0

import ccxt

class OKXEngine:
    def __init__(self, api_key: str, secret: str, passphrase: str, testnet: bool = True):
        self.ex = ccxt.okx({
            "apiKey": api_key,
            "secret": secret,
            "password": passphrase,
            "sandbox": testnet,
        })

    def fetch_ohlcv(self, symbol: str, timeframe: str = "1h", limit: int = 200):
        import pandas as pd
        data = self.ex.fetch_ohlcv(symbol, timeframe, limit=limit)
        df = pd.DataFrame(data, columns=["ts", "open", "high", "low", "close", "vol"])
        df["ts"] = pd.to_datetime(df["ts"], unit="ms")
        return df.set_index("ts")

    def market_buy(self, symbol: str, usdt_amount: float):
        ticker = self.ex.fetch_ticker(symbol)
        qty = usdt_amount / ticker["last"]
        return self.ex.create_market_buy_order(symbol, qty)

    def market_sell(self, symbol: str, qty: float):
        return self.ex.create_market_sell_order(symbol, qty)

from PySide6.QtWidgets import QMainWindow, QLabel, QHBoxLayout, QWidget
from PySide6.QtCore import QTimer
from PySide6.QtGui import QColor

class Dashboard(QMainWindow):
    def __init__(self, engine: OKXEngine):
        super().__init__()
        self.engine = engine
        self.setWindowTitle("OKX 量化仪表盘")
        self.resize(1200, 700)
        self.setStyleSheet("background:#0D1117;color:#E2E8F0;")
        self._build_ui()
        self._start_ticker()

    def _build_ui(self):
        central = QWidget()
        self.setCentralWidget(central)
        layout = QHBoxLayout(central)
        self.ticker_label = QLabel("BTC/USDT: --")
        self.ticker_label.setStyleSheet("font-size:24px;font-weight:bold;color:#22C55E;")
        layout.addWidget(self.ticker_label)

    def _start_ticker(self):
        self.timer = QTimer(self)
        self.timer.timeout.connect(self._update_ticker)
        self.timer.start(2000)  # 2秒刷新

    def _update_ticker(self):
        try:
            ticker = self.engine.ex.fetch_ticker("BTC/USDT")
            price = ticker["last"]
            change = ticker["percentage"]
            color = "#22C55E" if change >= 0 else "#EF4444"
            self.ticker_label.setText(
                f"BTC/USDT: ${price:,.2f}  {change:+.2f}%"
            )
            self.ticker_label.setStyleSheet(f"font-size:24px;font-weight:bold;color:{color};")
        except Exception as e:
            print(f"Ticker error: {e}")