【原】aitrader 2.0架構(gòu)思考與設(shè)計(jì)，附代碼

AI量化實(shí)驗(yàn)室 2025-06-23 發(fā)布于北京

展開(kāi)全文

原創(chuàng)內(nèi)容第926篇，專注智能量化投資、個(gè)人成長(zhǎng)與財(cái)富自由。

從主動(dòng)管理的投資，也就是希望通過(guò)主動(dòng)操作，獲得超額收益，應(yīng)該是廣義的投資者普遍認(rèn)為的剛需。

這里對(duì)比的被動(dòng)型投資，比如定投，風(fēng)險(xiǎn)平價(jià)等，面向長(zhǎng)期的視角，所以標(biāo)的物是有很多的歷史，有限個(gè)。

主動(dòng)管理的標(biāo)的物，更傾向于面向全市場(chǎng)。過(guò)去的歷史和很長(zhǎng)的未來(lái)，在波段操作者眼中，不重要。他們只關(guān)心明天、下周，下個(gè)月會(huì)不會(huì)漲。

大浪淘金，難度很高，但這就是大家希望量化，AI量化能夠幫忙做到的事情。

的確，全市場(chǎng)分析，選股是計(jì)算機(jī)的優(yōu)勢(shì)。然后按照每個(gè)人的投資偏好，比如風(fēng)險(xiǎn)收益比，比如回撤控制，比如分散的要求去構(gòu)建個(gè)性化的投資組合。

我們常用的“輪動(dòng)策略”，比如按多因子加權(quán)得分排序，多是動(dòng)量或趨勢(shì)。這里有一個(gè)隱含假設(shè)就是“低相關(guān)性”。就是一個(gè)趨勢(shì)強(qiáng)的時(shí)候，另一個(gè)趨勢(shì)弱。否則二者之間的動(dòng)量是無(wú)法直接比的。一個(gè)動(dòng)量是10%，另一個(gè)動(dòng)量是15%，是不是后者就一定強(qiáng)？不一定，有可能后者的波動(dòng)比較大。只有二者低相關(guān)性明顯，二者才可以這么比較。

比如如下這個(gè)年化590%的策略：

更合理的方式是：給每個(gè)標(biāo)的，獨(dú)立綜合評(píng)分。

就像給一個(gè)年級(jí)的學(xué)生綜合評(píng)分，會(huì)考查多個(gè)學(xué)科成績(jī)，來(lái)預(yù)判未來(lái)。只是這個(gè)綜合分在投資市場(chǎng)可以一天一評(píng)。

然后理想的投資組合就是選擇評(píng)分高的持有。

那你也許會(huì)問(wèn)，那不是所有人的組合都一樣？不一樣。

評(píng)分也許一樣，但會(huì)根據(jù)你自己的偏好，比如你要控制波動(dòng)，那么高波動(dòng)的標(biāo)的倉(cāng)位就不能太高。比如你要分散，要控制行業(yè)集中度等等。還有持有的周期和標(biāo)的數(shù)量，這樣基本就是千人千面了。

# 加載模塊from functools import partialimport polars as plfrom alpha.dataset import process_drop_na, process_cs_normfrom alpha.dataset.datasets.alpha_158 import Alpha158from trader.constant import Intervalfrom alpha import AlphaLab, AlphaDataset# 創(chuàng)建數(shù)據(jù)中心lab: AlphaLab = AlphaLab("./lab/etfs")# 設(shè)置任務(wù)參數(shù)name = "etfs"index_symbol: str = "000300.SSE"start: str = "2008-01-01"end: str = "2023-12-31"interval: Interval = Interval.DAILYextended_days: int = 100# 加載所有成分股代碼component_symbols: list[str] = lab.load_component_symbols(index_symbol, start, end)# 加載模塊# 加載成分股數(shù)據(jù)df: pl.DataFrame = lab.load_bar_df(component_symbols, interval, start, end, extended_days)# 創(chuàng)建數(shù)據(jù)集對(duì)象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"),)# 添加數(shù)據(jù)預(yù)處理器dataset.add_processor("learn", partial(process_drop_na, names=["label"]))dataset.add_processor("learn", partial(process_cs_norm, names=["label"], method="zscore"))# 收集指數(shù)成分過(guò)濾器filters: dict[str, list[str]] = lab.load_component_filters(index_symbol, start, end)# 準(zhǔn)備特征和標(biāo)簽數(shù)據(jù)dataset.prepare_data(filters, max_workers=3)# 保存到文件緩存lab.save_dataset(name, dataset)# 加載模塊import numpy as npfrom alpha import Segment, AlphaDataset, AlphaModelfrom alpha.model.models.lgb_model import LgbModel# 從文件緩存加載dataset: AlphaDataset = lab.load_dataset(name)# 創(chuàng)建模型對(duì)象model: AlphaModel = LgbModel(seed=42)# 使用數(shù)據(jù)集訓(xùn)練模型model.fit(dataset)# 保存模型lab.save_model(name, model)# 用模型在測(cè)試集上預(yù)測(cè)pre: np.ndarray = model.predict(dataset, Segment.TEST)# 加載測(cè)試集數(shù)據(jù)df_t: pl.DataFrame = dataset.fetch_infer(Segment.TEST)# 合并預(yù)測(cè)信號(hào)列df_t = df_t.with_columns(pl.Series(pre).alias("signal"))# 提取信號(hào)數(shù)據(jù)signal: pl.DataFrame = df_t["datetime", "vt_symbol", "signal"]# 加載模塊import importlibfrom datetime import datetimefrom alpha.strategy import BacktestingEngineimport alpha.strategy.strategies.equity_demo_strategy as equity_demo_strategy# 重載策略類importlib.reload(equity_demo_strategy)EquityDemoStrategy = equity_demo_strategy.EquityDemoStrategy# 從文件加載信號(hào)數(shù)據(jù)signal = lab.load_signal(name)# 創(chuàng)建回測(cè)引擎對(duì)象engine = BacktestingEngine(lab)# 設(shè)置回測(cè)參數(shù)engine.set_parameters(    vt_symbols=component_symbols,    interval=Interval.DAILY,    start=datetime(2017, 1, 1),    end=datetime(2020, 8, 1),    capital=100000000)# 添加策略實(shí)例setting = {"top_k": 30, "n_drop": 3, "hold_thresh": 3}engine.add_strategy(EquityDemoStrategy, setting, signal)# 執(zhí)行回測(cè)任務(wù)engine.load_data()engine.run_backtesting()engine.calculate_result()engine.calculate_statistics()engine.show_chart()