Building and Backtesting an Alpha158 Lasso Stock Signal
Summary
This workflow demonstrates an end-to-end daily equity modeling process using CSI 300 constituents and vn.py’s AlphaLab tools. It loads constituent histories, builds an Alpha158 dataset, and divides observations into training, validation, and test periods. Preprocessing drops missing labels and applies cross-sectional z-score normalization during learning, while inference fills missing values with zero. The dataset is then used to train a Lasso model and produce test-period predictions as a signal.
The example evaluates feature and signal performance, saves the dataset, model, and predictions, and passes the signal to an equity backtesting engine. The backtest uses a top-ranked selection and holding configuration, then calculates statistics and compares performance with the index benchmark. These steps illustrate a research pipeline, not evidence that the model is profitable: the excerpt gives no reported results, and its particular date splits, features, portfolio settings, and data handling choices need independent scrutiny. It does not discuss transaction costs or robustness across alternative samples.
Key ideas
- The workflow builds an Alpha158 dataset from daily index constituent data and divides it into training, validation, and test periods.
- Learning preprocessing removes missing labels and applies cross-sectional z-score normalization.
- A Lasso model generates predictions for the test segment, which are converted into a signal series.
- The signal is evaluated and passed to a portfolio backtesting engine for historical simulation.
- The example reports no performance figures and does not describe transaction cost modeling.
Tags
Full text
# 准备数据
# 准备数据
```python
# 过滤Alphalens的warning
import warnings
warnings.filterwarnings("ignore", category=FutureWarning)
```
```python
# 加载模块
import polars as pl
from vnpy.trader.constant import Interval
from vnpy.alpha import AlphaLab
```
```python
# 创建数据中心
lab: AlphaLab = AlphaLab("./lab/csi300")
```
```python
# 设置任务参数
name = "300_lasso"
index_symbol: str = "000300.SSE"
start: str = "2008-01-01"
end: str = "2023-12-31"
interval: Interval = Interval.DAILY
extended_days: int = 100
```
```python
# 加载所有成分股代码
component_symbols: list[str] = lab.load_component_symbols(index_symbol, start, end)
```
# 特征计算
```python
# 加载模块
from functools import partial
from vnpy.trader.constant import Interval
from vnpy.alpha.dataset import (
AlphaDataset,
process_drop_na,
process_cs_norm,
process_fill_na
)
from vnpy.alpha.dataset.datasets.alpha_158 import Alpha158
```
```python
# 加载成分股数据
df: pl.DataFrame = lab.load_bar_df(component_symbols, interval, start, end, extended_days)
```
```python
df
```
```python
# 创建数据集对象
dataset: AlphaDataset = Alpha158(
df,
train_period = ("2008-01-01", "2014-12-31"),
valid_period = ("2015-01-01", "2016-12-31"),
test_period = ("2017-01-01", "2020-8-31"),
)
```
```python
# 添加数据预处理器
dataset.add_processor("learn", partial(process_drop_na, names=["label"]))
dataset.add_processor("learn", partial(process_cs_norm, names=["label"], method="zscore"))
dataset.add_processor("infer", partial(process_fill_na, fill_value=0))
```
```python
# 收集指数成分过滤器
filters: dict[str, list[str]] = lab.load_component_filters(index_symbol, start, end)
```
```python
# 准备特征和标签数据
dataset.prepare_data(filters, max_workers=3)
```
```python
# 数据预处理
dataset.process_data()
```
```python
# 特征表现分析
dataset.show_feature_performance("rsv_5")
```
```python
# 保存到文件缓存
lab.save_dataset(name, dataset)
```
# 模型训练
```python
# 加载模块
import numpy as np
from vnpy.alpha import Segment, AlphaDataset, AlphaModel
from vnpy.alpha.model.models.lasso_model import LassoModel
```
```python
# 从文件缓存加载
dataset: AlphaDataset = lab.load_dataset(name)
```
```python
# 创建模型对象
model: AlphaModel = LassoModel()
```
```python
# 使用数据集训练模型
model.fit(dataset)
```
```python
# 查看模型细节
model.detail()
```
```python
# 保存模型
lab.save_model(name, model)
```
# 预测信号
```python
model: AlphaModel = lab.load_model(name)
```
```python
# 用模型在测试集上预测
pre: np.ndarray = model.predict(dataset, Segment.TEST)
# 加载测试集数据
df_t: pl.DataFrame = dataset.fetch_infer(Segment.TEST)
# 合并预测信号列
df_t = df_t.with_columns(pl.Series(pre).alias("signal"))
# 提取信号数据
signal: pl.DataFrame = df_t["datetime", "vt_symbol", "signal"]
```
```python
# 检查信号绩效
dataset.show_signal_performance(signal)
```
```python
# 保存信号数据
lab.save_signal(name, signal)
```
# 策略回测
```python
# 加载模块
import importlib
from datetime import datetime
from vnpy.alpha.strategy import BacktestingEngine
import vnpy.alpha.strategy.strategies.equity_demo_strategy as equity_demo_strategy
```
```python
# 重载策略类
importlib.reload(equity_demo_strategy)
EquityDemoStrategy = equity_demo_strategy.EquityDemoStrategy
```
```python
# 从文件加载信号数据
signal = lab.load_signal(name)
```
```python
# 创建回测引擎对象
engine = BacktestingEngine(lab)
# 设置回测参数
engine.set_parameters(
vt_symbols=component_symbols,
interval=Interval.DAILY,
start=datetime(2017, 1, 1),
end=datetime(2020, 8, 1),
capital=100000000
)
# 添加策略实例
setting = {"top_k": 30, "n_drop": 3, "hold_thresh": 3}
engine.add_strategy(EquityDemoStrategy, setting, signal)
```
```python
# 执行回测任务
engine.load_data()
engine.run_backtesting()
engine.calculate_result()
engine.calculate_statistics()
engine.show_chart()
```
```python
# 显示超额收益分析结果
engine.show_performance(benchmark_symbol=index_symbol)
```
```python
```Shown in full with attribution under the source's licence. Licence: MIT
This summary was written by Stratmill's research agent from the original; it is not a copy of the source.