Skip to content
Published on

AI 金融与量化交易：从 FinBERT、强化学习到回测

🇰🇷KO🇺🇸EN🇯🇵JA🇨🇳ZH
필사 모드
分享
Authors

1. 金融数据采集与预处理

量化交易的起点是数据。除了 OHLCV（开盘价·最高价·最低价·收盘价·成交量）之外，订单簿快照、逐笔数据（tick data）、新闻与卫星图像等另类数据的应用范围也在不断扩展。

用 yfinance 下载股票数据

import yfinance as yf
import pandas as pd

# 批量下载多个标的的日线 OHLCV 数据
tickers = ["AAPL", "MSFT", "GOOGL", "NVDA"]
df = yf.download(tickers, start="2020-01-01", end="2026-01-01", auto_adjust=True)

# 将 MultiIndex 转换为按标的划分的单一 DataFrame
close = df["Close"]
volume = df["Volume"]

# 缺失值处理：前向填充后删除首个日期之前的行
close = close.ffill().dropna()

print(close.tail())

用 ccxt 采集加密货币订单簿

import ccxt
import time

exchange = ccxt.binance()

symbol = "BTC/USDT"
orderbook = exchange.fetch_order_book(symbol, limit=20)

bids = orderbook["bids"][:5]   # [价格, 数量] 前 5 档买单
asks = orderbook["asks"][:5]   # [价格, 数量] 前 5 档卖单

mid_price = (bids[0][0] + asks[0][0]) / 2
spread_bps = (asks[0][0] - bids[0][0]) / mid_price * 10000

print(f"Mid: {mid_price:.2f}, Spread: {spread_bps:.2f} bps")

另类数据：新闻标题采集

新闻、社交媒体数据能提供结构化数据无法捕捉的自然语言阿尔法（Natural Language Alpha）。

import requests
from datetime import datetime, timedelta

API_KEY = "YOUR_NEWSAPI_KEY"
yesterday = (datetime.today() - timedelta(days=1)).strftime("%Y-%m-%d")

url = (
    f"https://newsapi.org/v2/everything"
    f"?q=NVIDIA+earnings&from={yesterday}&sortBy=publishedAt"
    f"&language=en&apiKey={API_KEY}"
)
resp = requests.get(url).json()
headlines = [art["title"] for art in resp.get("articles", [])]
print(headlines[:5])

2. 技术分析自动化

借助 TA-Lib 与 pandas-ta，只需一行 Python 代码就能计算数百种技术指标。

RSI / MACD 计算

import talib
import numpy as np
import yfinance as yf

df = yf.download("AAPL", start="2023-01-01", end="2026-01-01", auto_adjust=True)
close = df["Close"].squeeze().values.astype(float)

# RSI（14 日）
rsi = talib.RSI(close, timeperiod=14)

# MACD
macd, signal, hist = talib.MACD(close, fastperiod=12, slowperiod=26, signalperiod=9)

# pandas-ta 方式（无需 TA-Lib）
import pandas_ta as ta
df_ta = df["Close"].squeeze().to_frame("close")
df_ta.ta.rsi(length=14, append=True)
df_ta.ta.macd(fast=12, slow=26, signal=9, append=True)
print(df_ta.tail())

形态识别：蜡烛图

# TA-Lib 蜡烛图形态识别示例
open_p  = df["Open"].squeeze().values.astype(float)
high_p  = df["High"].squeeze().values.astype(float)
low_p   = df["Low"].squeeze().values.astype(float)
close_p = df["Close"].squeeze().values.astype(float)

hammer       = talib.CDLHAMMER(open_p, high_p, low_p, close_p)
engulfing    = talib.CDLENGULFING(open_p, high_p, low_p, close_p)
morning_star = talib.CDLMORNINGSTAR(open_p, high_p, low_p, close_p)

# 100 或 -100（检测到强势/弱势形态），0 为不适用
print("Hammer signals:", (hammer != 0).sum())

3. ML 交易策略：XGBoost 阿尔法因子

特征工程

import pandas as pd
import numpy as np
import yfinance as yf
import pandas_ta as ta

df = yf.download("SPY", start="2018-01-01", end="2026-01-01", auto_adjust=True)
df.columns = df.columns.droplevel(1) if df.columns.nlevels > 1 else df.columns
df.columns = [c.lower() for c in df.columns]

# 收益率特征
df["ret_1d"]  = df["close"].pct_change(1)
df["ret_5d"]  = df["close"].pct_change(5)
df["ret_20d"] = df["close"].pct_change(20)

# 波动率特征
df["vol_20d"] = df["ret_1d"].rolling(20).std()

# 技术指标特征
df.ta.rsi(length=14, append=True)
df.ta.macd(fast=12, slow=26, signal=9, append=True)
df.ta.bbands(length=20, append=True)

# 成交量特征
df["vol_ratio"] = df["volume"] / df["volume"].rolling(20).mean()

# 目标：未来 5 日收益率符号（1：上涨，0：下跌）
df["target"] = (df["close"].pct_change(5).shift(-5) > 0).astype(int)

df.dropna(inplace=True)
print(df.shape)

用 Walk-Forward 验证训练 XGBoost

from xgboost import XGBClassifier
from sklearn.metrics import accuracy_score, roc_auc_score
import warnings
warnings.filterwarnings("ignore")

feature_cols = [
    "ret_1d", "ret_5d", "ret_20d", "vol_20d",
    "RSI_14", "MACD_12_26_9", "MACDs_12_26_9",
    "BBL_20_2.0", "BBM_20_2.0", "BBU_20_2.0",
    "vol_ratio"
]
target_col = "target"

# Walk-Forward：1 年训练 → 3 个月测试滑动
results = []
train_years = 2
test_months = 3

dates = df.index
start_year = dates[0].year + train_years

for year in range(start_year, 2026):
    for q in range(1, 5):
        train_end = pd.Timestamp(f"{year}-{(q-1)*3+1:02d}-01") if q > 1 else pd.Timestamp(f"{year}-01-01")
        test_start = train_end
        test_end   = test_start + pd.DateOffset(months=test_months)

        train_df = df[df.index < test_start].tail(504)  # 约 2 年的数据量
        test_df  = df[(df.index >= test_start) & (df.index < test_end)]

        if len(train_df) < 100 or len(test_df) < 10:
            continue

        X_train, y_train = train_df[feature_cols], train_df[target_col]
        X_test,  y_test  = test_df[feature_cols],  test_df[target_col]

        model = XGBClassifier(n_estimators=200, max_depth=4,
                              learning_rate=0.05, subsample=0.8,
                              eval_metric="logloss", random_state=42)
        model.fit(X_train, y_train)

        preds = model.predict(X_test)
        proba = model.predict_proba(X_test)[:, 1]

        acc = accuracy_score(y_test, preds)
        auc = roc_auc_score(y_test, proba)
        results.append({"period": str(test_start.date()), "acc": acc, "auc": auc})

result_df = pd.DataFrame(results)
print(result_df.tail(8))
print(f"\n平均 AUC: {result_df['auc'].mean():.4f}")

4. 金融深度学习：LSTM 与 Temporal Fusion Transformer

LSTM 股价预测

import numpy as np
import torch
import torch.nn as nn
from sklearn.preprocessing import MinMaxScaler

# 准备数据（对收盘价做归一化）
scaler = MinMaxScaler()
scaled = scaler.fit_transform(df[["close"]].values)

SEQ_LEN = 60

def make_sequences(data, seq_len):
    X, y = [], []
    for i in range(len(data) - seq_len):
        X.append(data[i:i+seq_len])
        y.append(data[i+seq_len])
    return np.array(X), np.array(y)

X, y = make_sequences(scaled, SEQ_LEN)
split = int(len(X) * 0.8)
X_train, X_test = X[:split], X[split:]
y_train, y_test = y[:split], y[split:]

X_train_t = torch.tensor(X_train, dtype=torch.float32)
y_train_t = torch.tensor(y_train, dtype=torch.float32)

class LSTMModel(nn.Module):
    def __init__(self, input_size=1, hidden_size=64, num_layers=2):
        super().__init__()
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True, dropout=0.2)
        self.fc   = nn.Linear(hidden_size, 1)

    def forward(self, x):
        out, _ = self.lstm(x)
        return self.fc(out[:, -1, :])

model = LSTMModel()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
criterion = nn.MSELoss()

for epoch in range(30):
    model.train()
    pred = model(X_train_t)
    loss = criterion(pred, y_train_t)
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()
    if (epoch + 1) % 10 == 0:
        print(f"Epoch {epoch+1}, Loss: {loss.item():.6f}")

FinRL 强化学习交易智能体（配置示例）

FinRL 是基于 OpenAI Gym 环境、用于训练股票交易强化学习智能体的框架。

# pip install finrl
from finrl.meta.env_stock_trading.env_stocktrading import StockTradingEnv
from finrl.agents.stablebaselines3.models import DRLAgent
import pandas as pd

# 已预处理为 FinRL 格式的金融数据
# 必需列：date, tic, open, high, low, close, volume + 技术指标
processed_df = pd.read_csv("processed_stock_data.csv")

# 环境设置
env_kwargs = {
    "hmax": 100,               # 最大持仓股数
    "initial_amount": 100000,  # 初始资金（美元）
    "buy_cost_pct": 0.001,     # 交易手续费 0.1%
    "sell_cost_pct": 0.001,
    "reward_scaling": 1e-4,
    "state_space": 181,
    "action_space": 30,
    "tech_indicator_list": ["macd", "rsi_30", "cci_30", "dx_30"],
}

train_env = StockTradingEnv(df=processed_df, **env_kwargs)

# 训练 PPO 智能体
agent = DRLAgent(env=train_env)
model_ppo = agent.get_model("ppo")
trained_ppo = agent.train_model(
    model=model_ppo,
    tb_log_name="ppo_stock",
    total_timesteps=50000
)

5. LLM for Finance：FinBERT 情感分析

FinBERT 是专为金融新闻、财报电话会议微调的 BERT 预训练模型，执行 Positive / Negative / Neutral 三分类。

from transformers import BertTokenizer, BertForSequenceClassification
import torch
import torch.nn.functional as F

model_name = "ProsusAI/finbert"
tokenizer  = BertTokenizer.from_pretrained(model_name)
model      = BertForSequenceClassification.from_pretrained(model_name)
model.eval()

def finbert_sentiment(texts):
    inputs = tokenizer(texts, padding=True, truncation=True,
                       max_length=512, return_tensors="pt")
    with torch.no_grad():
        logits = model(**inputs).logits
    probs = F.softmax(logits, dim=-1).numpy()
    labels = ["positive", "negative", "neutral"]
    return [
        {"text": t, "label": labels[p.argmax()], "score": float(p.max())}
        for t, p in zip(texts, probs)
    ]

headlines = [
    "NVIDIA beats Q4 earnings estimates by 15%, raises guidance",
    "Fed signals higher-for-longer rates amid sticky inflation",
    "Apple reports record services revenue despite iPhone slowdown",
]
results = finbert_sentiment(headlines)
for r in results:
    print(f"[{r['label'].upper():8s}] {r['score']:.3f} | {r['text']}")

财报摘要：数字/指引与文本语气分离分析

先把数字（EPS、营收）与指引数值单独解析出来，再用 FinBERT 对管理层的文本发言给出情感分数。把这两类信号结合起来，能生成比单纯文本分析更精确的阿尔法。

6. 风险管理：VaR、CVaR 与凯利公式

VaR / CVaR 计算

import numpy as np
import scipy.stats as stats

def calculate_var_cvar(returns, confidence=0.95):
    """
    历史模拟法 VaR/CVaR 计算
    returns：日收益率数组
    """
    sorted_returns = np.sort(returns)
    index = int((1 - confidence) * len(sorted_returns))
    var  = -sorted_returns[index]
    cvar = -sorted_returns[:index].mean()
    return var, cvar

daily_returns = df["close"].pct_change().dropna().values
var_95, cvar_95 = calculate_var_cvar(daily_returns, 0.95)
var_99, cvar_99 = calculate_var_cvar(daily_returns, 0.99)

print(f"VaR  95%: {var_95:.4f} ({var_95*100:.2f}%)")
print(f"CVaR 95%: {cvar_95:.4f} ({cvar_95*100:.2f}%)")
print(f"VaR  99%: {var_99:.4f} ({var_99*100:.2f}%)")
print(f"CVaR 99%: {cvar_99:.4f} ({cvar_99*100:.2f}%)")

主要风险指标对比表

指标公式特点局限
夏普比率(Rp - Rf) / σp标准化的风险调整收益对上行波动也施加惩罚
索提诺比率(Rp - Rf) / σd仅对下行波动施加惩罚分母计算不直观
最大回撤（MDD）相对最高点的最大损失捕捉极端损失未反映恢复期
VaR 95%下侧 5% 损失分位数监管标准低估尾部风险
CVaR 95%超过 VaR 部分损失的期望值反映尾部风险对参数估计敏感
卡玛比率CAGR / MDD回撤对比下的成长性不适合短期分析

凯利公式仓位管理

def kelly_fraction(win_rate, win_loss_ratio):
    """
    f* = W - (1-W)/R
    W：胜率，R：平均盈亏比
    """
    return win_rate - (1 - win_rate) / win_loss_ratio

# 示例：胜率 55%，盈亏比 1.5
f_full = kelly_fraction(0.55, 1.5)
f_half = f_full * 0.5   # fractional Kelly（降低波动性）

print(f"Full Kelly: {f_full:.2%}")
print(f"Half Kelly: {f_half:.2%}")

之所以采用 fractional Kelly（通常为 0.25～0.5 倍）：实际胜率估计的误差很大，而 full Kelly 会低估破产概率，因此需要施加安全边际。

7. 回测：用 Vectorbt 验证策略

用 Vectorbt 做多空策略回测

import vectorbt as vbt
import pandas as pd
import yfinance as yf

# 加载数据
price = yf.download("SPY", start="2018-01-01", end="2026-01-01",
                    auto_adjust=True)["Close"].squeeze()

# 生成均线交叉信号
fast_ma = vbt.MA.run(price, 20)
slow_ma = vbt.MA.run(price, 60)

entries = fast_ma.ma_crossed_above(slow_ma)
exits   = fast_ma.ma_crossed_below(slow_ma)

# 组合模拟
portfolio = vbt.Portfolio.from_signals(
    price,
    entries,
    exits,
    init_cash=100_000,
    fees=0.001,        # 手续费 0.1%
    slippage=0.001,    # 滑点 0.1%
    freq="D",
)

# 输出绩效指标
stats = portfolio.stats()
print(stats[["Total Return [%]", "Sharpe Ratio", "Max Drawdown [%]",
             "Win Rate [%]", "Profit Factor"]])

过拟合防范清单

回测出现过于乐观的结果时，必须检查以下项目：

偏差种类原因应对方法
前视偏差用未来数据计算当前信号检查 shift(-1)，复核特征生成时点
幸存者偏差剔除已退市标的使用完整的全量标的历史数据
优化偏差样本内参数过拟合Walk-forward、WFO 测试
忽视市场冲击忽视大额订单对价格的影响滑点模型、成交量限制
低估交易成本未反映实际点差/手续费设置贴近实际的手续费与滑点

测验

Q1. 为什么 walk-forward 验证比 k-fold 交叉验证更适合金融时间序列？

答案：时间序列数据存在时间依赖性（temporal dependency），若像 k-fold 那样随机切分，会让未来数据混入训练集，产生前视偏差。

解析：Walk-forward 验证始终用过去的数据训练、用未来的数据测试，保持了时间顺序。而 k-fold 会让某些折（fold）内混入晚于测试时点的收益率作为训练数据，模型因此像是"记住"了未来信息，导致表现被夸大。金融时间序列存在自相关（autocorrelation）与体制转换（regime shift），因此必须使用保留时间顺序的验证方式。

Q2. 夏普比率的局限性，以及索提诺比率更适用的情形是？

答案：夏普比率把上行和下行波动同等地视为惩罚项，因此会对那些因收益高而波动大的策略给出不公平的低评分。

解析：索提诺比率把分母替换为下方标准差（downside deviation），只对投资者真正厌恶的、损失方向的波动施加惩罚。对于卖出期权、动量策略这类收益分布非对称、右尾较厚的策略，索提诺比率更适合用来评估。

Q3. 前视偏差（look-ahead bias）是如何让回测结果被乐观扭曲的？

答案：生成信号时，把那个时点其实还不存在的未来数据（当日收盘价、下一季度财报等）带入了特征或标签的计算，导致模型呈现出远高于实际水平的预测准确率。

解析：举例来说，如果用当日收盘价算出的 RSI 去触发同一天开盘价的成交信号，等于提前用到了实际交易中根本不可能知道的信息。常见原因还包括：pandas 里忘记处理 shift(-n)、rolling 统计的 min_periods 设置错误、指数移动平均把未来信息向前泄漏等。

Q4. Kelly Criterion 是仓位管理数学最优解的依据，以及为什么要用 fractional Kelly？

答案：凯利公式从数学上推导出能使长期期望对数收益（expected log return）最大化的下注比例。在 f* = W - (1-W)/R 这个公式中，W 是胜率，R 是盈亏比。

解析：凯利公式使几何平均增长率最大化，理论上能让资产以最快速度长期增值。但如果胜率、盈亏比的估计误差较大，过度的杠杆可能引发严重回撤。为弥补这一点，实务中通常使用相当于 full Kelly 的 25~50% 的 fractional Kelly，以降低破产风险、管理波动性。

Q5. 用 LLM 分析财报情感时，为什么要把数字/指引与文本语气分开分析？

答案：管理层经常一边公布亮眼的数字，一边对未来展望（指引）给出保守的说法；反过来，也常把糟糕的业绩用积极的措辞包装。混在一起分析时，信号就会被稀释。

解析：财报中包含三种信号：(1) EPS、营收等定量数字，(2) 未来指引数值，(3) 管理层发言的文本语气。FinBERT 等 LLM 擅长捕捉文本语气，但像「低于预期 10%」这种嵌入了数字的句子，文本模型可能会误判。把数字解析（regex / NLP 结构化）与文本情感分析分离开来再做集成，就能得到更强的自然语言阿尔法。

Discuss on Twitter • View on GitHub