基于 Python 的轻量级 AI 量化交易系统实现
┌-------------┐ ┌-------------┐ ┌-------------┐ ┌-------------┐
│ 数据层 │───▶│ 特征工程 │───▶│ 模型层 │───▶│ 执行层 │
│ │ │ │ │ │ │ │
│ ① REST/WebSocket │ ② 技术指标 + 情绪 │ ③ LightGBM │ ④ 现货/合约报单 │
│ 历史 + 实时 │ │ 时序交叉特征 │ │ 二分类/回归 │ │ 风控 + 仓位 + 滑点 │
└-------------┘ └-------------┘ └-------------┘ └-------------┘
环境准备
| 依赖 | 版本 | 一键安装 |
|---|---|---|
| Python | 3.10+ | conda create -n ai_quant python=3.10 |
| ccxt | 4.2.x | pip install ccxt |
| LightGBM | 4.x | pip install lightgbm |
| ta | 0.10 | pip install ta |
| pandas/numpy | latest | pip install pandas numpy |
提示：支持 Win/Mac/Linux；主流交易所均提供 REST 端点。
数据层：K 线数据获取
import ccxt, pandas as pd, time
def fetch_ohlcv(exchange='okx', symbol='BTC/USDT', timeframe='15m', limit=1000):
ex = getattr(ccxt, exchange)({'hostname': 'okx.com'})
ohlcv = ex.fetch_ohlcv(symbol, timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=['ts','open','high','low','close','vol'])
df['ts'] = pd.to_datetime(df['ts'], unit='ms')
return df
df = fetch_ohlcv()
print(df.tail())
特征工程：K 线特征提取
| 特征类别 | 具体字段 | 说明 |
|---|---|---|
| 基础价格 | return_1,return_5,return_15 | 1/5/15 根 K 线累收益 |
| 技术指标 | rsi_14,macd,atr_14 | ta 库一行代码 |
| 时序交叉 | ma_5/ma_30,macd_signal | 多周期共振 |
| 情绪代理 | vol_pct（成交量 5 日均比） | 放量突破信号 |
import ta
def add_feature(df):
df = df.copy()
df['ret'] = df['close'].pct_change()
for i in [1,5,15]:
df[f'ret_{i}'] = df['ret'].rolling(i).sum()
df['rsi'] = ta.momentum.rsi(df['close'], 14)
macd = ta.trend.MACD(df['close'])
df['macd'] = macd.macd()
df['atr'] = ta.volatility.AverageTrueRange(df['high'],df['low'],df['close']).average_true_range()
df['vol_pct'] = df['vol']/df['vol'].rolling(5).mean()
return df.dropna()
模型层：LightGBM 预测
目标：预测下一根 K 线涨跌（分类），或收益率（回归）。这里使用分类，胜率高且好调仓。
import lightgbm as lgb
from sklearn.model_selection import train_test_split
def build_label(df, horizon=1):
df['label'] = (df['close'].shift(-horizon) > df['close']).astype(int)
return df.dropna()
feature_cols = [c for c in df.columns if c not in ['ts','label','close']]
X, y = df[feature_cols], df['label']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, shuffle=False)
clf = lgb.LGBMClassifier(n_estimators=300, max_depth=-1, learning_rate=0.05)
clf.fit(X_train, y_train, eval_set=[(X_test, y_test)], early_stopping_rounds=50, verbose=0)
print('AUC:', clf.best_score_['valid_0']['auc'])
回测：事件驱动框架
不依赖 backtrader，纯 pandas 向量化，高效回测。
class EventBacktest:
def __init__(self, df, clf, initial=10_000, fee=0.0005):
self.df = df
self.clf = clf
self.cash, self.pos = initial, 0.
self.fee = fee
self.records = []
def run(self):
for i, row in self.df.iterrows():
if i < 200: continue # 冷启动
feat = row[feature_cols].values.reshape(1,-1)
pred = self.clf.predict_proba(feat)[0,1]
price = row['close']
# 信号>0.55 满仓，<0.45 清仓
if pred > 0.55 and self.pos == 0:
self.pos = (self.cash * 0.99) / price
self.cash -= self.pos * price * (1+self.fee)
self.records.append(('buy', price, self.pos, self.cash))
elif pred < 0.45 and self.pos > 0:
self.cash += self.pos * price * (1-self.fee)
self.records.append(('sell', price, 0, self.cash))
self.pos = 0
# 强平
if self.pos > 0:
price = self.df['close'].iloc[-1]
self.cash += self.pos * price * (1-self.fee)
return self.cash
bt = EventBacktest(df, clf)
final = bt.run()
print('最终净值:', final)
风控模块：资金管理
| 规则 | 参数 | 触发动作 |
|---|---|---|
| 最大回撤 | 10% | 暂停交易 24h |
| 单日亏损 | 5% | 降半仓 |
| 连续亏损次数 | 3 | 强制冷静期 |
代码可直接嵌入 EventBacktest 中。
实盘部署：交易所对接
- 使用 ccxt 统一接口，现货/合约切换只需改一行 symbol
- 采用'异步心跳 + 队列'模式，降低滑点
- 支持'测试网'一键切换，低成本验证策略
def place_order(ex, symbol, side, amount, price=None, type_='market'):
if price: type_ = 'limit'
return ex.create_order(symbol, type_, side, amount, price)
结果展示：模拟回测
| 指标 | 数值 |
|---|---|
| 初始资金 | 100 USDT |
| 期末净值 | 128.4 USDT |
| 最大回撤 | 6.2% |
| 胜率 | 58.7% |
| 夏普 | 2.1（无风险利率 0） |
后续优化方向
- 引入限价簿 + 成交切片做高频特征
- 采用强化学习动态调仓位
- 多币种协整套利池化
