量化交易进阶篇：非线性模型、另类数据与机构级 Alpha

import numpy as np
import pandas as pd
import lightgbm as lgb
from sklearn.metrics import spearmanr
 
np.random.seed(42)
# 500 天 × 50 只股票，3 个因子 + 噪声
days, stocks = 500, 50
dates = pd.date_range("2022-01-01", periods=days, freq="B")
 
records = []
for d in dates:
    f_value = np.random.randn(stocks)
    f_mom   = np.random.randn(stocks)
    f_liq   = np.random.randn(stocks)
    # 真实关系：非线性——动量仅在流动性好时有效
    ret_fwd = (
        0.02 * f_value
        + 0.03 * f_mom * (f_liq > 0).astype(float)
        + np.random.randn(stocks) * 0.1
    )
    for i in range(stocks):
        records.append({
            "date": d, "stock": f"s{i:02d}",
            "value": f_value[i], "mom": f_mom[i], "liq": f_liq[i],
            "ret_fwd": ret_fwd[i],
        })
 
df = pd.DataFrame(records)
features = ["value", "mom", "liq"]
 
# 走步验证：用过去 400 天训练，预测后 100 天
split = dates[400]
train = df[df["date"] < split]
test  = df[df["date"] >= split]
 
model = lgb.LGBMRegressor(
    n_estimators=200, learning_rate=0.05,
    max_depth=4, subsample=0.8, colsample_bytree=0.8,
    verbosity=-1,
)
model.fit(train[features], train["ret_fwd"])
pred = model.predict(test[features])
 
# 评价：每日截面 Spearman IC（预测排名 vs 真实收益排名）
test = test.copy()
test["pred"] = pred
ic_list = []
for d, g in test.groupby("date"):
    ic, _ = spearmanr(g["pred"], g["ret_fwd"])
    ic_list.append(ic)
print(f"样本外平均 IC: {np.nanmean(ic_list):.4f}")
print("特征重要性:", dict(zip(features, model.feature_importances_.round(2))))

import numpy as np
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset
 
np.random.seed(0)
torch.manual_seed(0)
 
# 构造序列：用过去 20 日收益预测下一日符号
T, seq_len = 3000, 20
rets = np.random.normal(0, 0.01, T)
# 嵌入微弱可预测性
for t in range(seq_len, T):
    rets[t] += 0.3 * rets[t - 1] + np.random.normal(0, 0.009)
 
X, y = [], []
for t in range(seq_len, T - 1):
    X.append(rets[t - seq_len : t])
    y.append(1 if rets[t + 1] > 0 else 0)
X = torch.tensor(np.array(X), dtype=torch.float32).unsqueeze(-1)
y = torch.tensor(y, dtype=torch.float32)
 
split = int(0.8 * len(y))
train_dl = DataLoader(TensorDataset(X[:split], y[:split]), batch_size=64, shuffle=True)
 
class LSTMClassifier(nn.Module):
    def __init__(self):
        super().__init__()
        self.lstm = nn.LSTM(1, 32, batch_first=True)
        self.fc = nn.Linear(32, 1)
    def forward(self, x):
        out, _ = self.lstm(x)
        return torch.sigmoid(self.fc(out[:, -1, :]))
 
model = LSTMClassifier()
opt = torch.optim.Adam(model.parameters(), lr=1e-3)
loss_fn = nn.BCELoss()
 
for epoch in range(5):
    for xb, yb in train_dl:
        opt.zero_grad()
        loss = loss_fn(model(xb).squeeze(), yb)
        loss.backward()
        opt.step()
 
model.eval()
with torch.no_grad():
    acc = ((model(X[split:]).squeeze() > 0.5) == y[split:]).float().mean()
print(f"样本外方向准确率: {acc:.3f}（微弱优于 50% 即可能有交易价值，需扣费验证）")

原始文本 → 时间戳对齐 → 实体链接(股票代码) → 情绪/主题/事件抽取 → 日频因子 → 与价量合并 → 模型

import numpy as np
import pandas as pd
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
 
np.random.seed(1)
n = 400
# 合成：正面词多的日子，次日收益略高
vocab_pos = ["增长", "超预期", "回购", "突破"]
vocab_neg = ["下滑", "调查", "亏损", "减持"]
 
texts, labels, fwd_ret = [], [], []
for i in range(n):
    pos = np.random.randint(0, 4)
    neg = np.random.randint(0, 4)
    words = vocab_pos[:pos] + vocab_neg[:neg]
    np.random.shuffle(words)
    texts.append(" ".join(words) if words else "平稳")
    sentiment = pos - neg
    labels.append(1 if sentiment > 0 else 0)
    fwd_ret.append(sentiment * 0.002 + np.random.normal(0, 0.01))
 
split = 300
pipe = Pipeline([
    ("tfidf", TfidfVectorizer(max_features=50)),
    ("clf", LogisticRegression(max_iter=500)),
])
pipe.fit(texts[:split], labels[:split])
proba = pipe.predict_proba(texts[split:])[:, 1]
 
# 情绪概率 vs 次日收益
ic = np.corrcoef(proba, fwd_ret[split:])[0, 1]
print(f"情绪概率与次日收益相关: {ic:.4f}")

import numpy as np
import pandas as pd
import lightgbm as lgb
from sklearn.metrics import spearmanr
 
def walk_forward_ic(df, features, train_days=400, test_days=20, embargo=5):
    """滚动训练 test_days，训练集与测试集之间留 embargo 天"""
    dates = sorted(df["date"].unique())
    ics = []
    for start in range(train_days, len(dates) - test_days, test_days):
        train_end = dates[start - embargo - 1]
        test_start = dates[start]
        test_end = dates[min(start + test_days - 1, len(dates) - 1)]
        tr = df[df["date"] <= train_end]
        te = df[(df["date"] >= test_start) & (df["date"] <= test_end)]
        if len(te) < 50:
            continue
        m = lgb.LGBMRegressor(n_estimators=100, verbosity=-1)
        m.fit(tr[features], tr["ret_fwd"])
        te = te.copy()
        te["pred"] = m.predict(te[features])
        for d, g in te.groupby("date"):
            ic, _ = spearmanr(g["pred"], g["ret_fwd"])
            ics.append(ic)
    return np.nanmean(ics), ics
 
# 复用上一章 df
mean_ic, all_ic = walk_forward_ic(df, features)
print(f"Walk-forward 平均 IC: {mean_ic:.4f}, 窗口数: {len(all_ic)}")

原理篇 → 模型篇（经典力学）→ 本文（非线性 + 另类数据）→ 专项深入