当前位置：首页 > news >正文

十年年化50%+的策略如何进化？兼容机器学习流程的量化策略开发，附python代码

news 2025/6/20 10:02:54

原创内容第922篇，专注智能量化投资、个人成长与财富自由。

咱们星球叫“智能量化实验室“，有一部分智能的工作，落在了AGI星球来实现。

但智能量化框架，还有深挖的空间。

区别于传统量化（backtrader，bt等）都是传统量化。也就是技术分析的”自动化“版本。

比如咱们的动量轮动，排序因子，还是止盈、止损规则。

当然这是有效的，也是可解释的——大家看策略榜的更新就知道了：

然而，这里还没有太多机器学习、深度学习等智能化相关的东西。

策略也不会自主学习和进化。

qlib框架是最接近AI驱动的开源量化框架，设计理念就是机器学习导向。

不过咱们之前也分析过，它不太兼容传统规则量化，而且表达式与自定义的数据存储格式绑定，不容易切换到自己的数据存储。vnpy.alpha在这方面处理得挺好。

import jsonimport shelveimport picklefrom pathlib import Pathfrom datetime import datetime, timedeltafrom collections import defaultdictfrom functools import lru_cache, partialimport numpy as npfrom loguru import loggerimport polars as plfrom constant import Intervalfrom dataset import to_datetime, AlphaDataset, process_drop_na, process_cs_norm, Segmentfrom dataset.datasets.alpha_158 import Alpha158from model import AlphaModelfrom model.models.lgb_model import LgbModelfrom strategy.strategies.equity_demo_strategy import EquityDemoStrategyclass AlphaLab:    def __init__(self, lab_path: str, data_path:str) -> None:        self.lab_path: Path = Path(lab_path)        self.data_path = Path(data_path)        self.dataset_path: Path = self.lab_path.joinpath("dataset")        self.model_path: Path = self.lab_path.joinpath("model")        self.signal_path: Path = self.lab_path.joinpath("signal")        # 创建文件夹        for path in [            self.lab_path,            self.dataset_path,            self.model_path,            self.signal_path        ]:            if not path.exists():                path.mkdir(parents=True)    def save_dataset(self, name: str, dataset: AlphaDataset) -> None:            """Save dataset"""            file_path: Path = self.dataset_path.joinpath(f"{name}.pkl")            with open(file_path, mode="wb") as f:                pickle.dump(dataset, f)    def load_dataset(self, name: str) -> AlphaDataset | None:        """Load dataset"""        file_path: Path = self.dataset_path.joinpath(f"{name}.pkl")        if not file_path.exists():            logger.error(f"Dataset file {name} does not exist")            return None        with open(file_path, mode="rb") as f:            dataset: AlphaDataset = pickle.load(f)            return dataset    def save_model(self, name: str, model: AlphaModel) -> None:        """Save model"""        file_path: Path = self.model_path.joinpath(f"{name}.pkl")        with open(file_path, mode="wb") as f:            pickle.dump(model, f)    def load_model(self, name: str) -> AlphaModel | None:        """Load model"""        file_path: Path = self.model_path.joinpath(f"{name}.pkl")        if not file_path.exists():            logger.error(f"Model file {name} does not exist")            return None        with open(file_path, mode="rb") as f:            model: AlphaModel = pickle.load(f)            return model    def remove_model(self, name: str) -> bool:        """Remove model"""        file_path: Path = self.model_path.joinpath(f"{name}.pkl")        if not file_path.exists():            logger.error(f"Model file {name} does not exist")            return False        file_path.unlink()        return True    def list_all_models(self) -> list[str]:        """List all models"""        return [file.stem for file in self.model_path.glob("*.pkl")]    def save_signal(self, name: str, signal: pl.DataFrame) -> None:        """Save signal"""        file_path: Path = self.signal_path.joinpath(f"{name}.parquet")        signal.write_parquet(file_path)    def load_signal(self, name: str) -> pl.DataFrame | None:        """Load signal"""        file_path: Path = self.signal_path.joinpath(f"{name}.parquet")        if not file_path.exists():            logger.error(f"Signal file {name} does not exist")            return None        return pl.read_parquet(file_path)    def remove_signal(self, name: str) -> bool:        """Remove signal"""        file_path: Path = self.signal_path.joinpath(f"{name}.parquet")        if not file_path.exists():            logger.error(f"Signal file {name} does not exist")            return False        file_path.unlink()        return True    def list_all_signals(self) -> list[str]:        """List all signals"""        return [file.stem for file in self.model_path.glob("*.parquet")]    def load_bar_df(self,symbols: list[str],                    start: datetime | str='20100101',                    end: datetime | str=datetime.now().strftime('%Y%m%d'),                    extended_days: int=20):        start = to_datetime(start) - timedelta(days=extended_days)        end = to_datetime(end) + timedelta(days=extended_days // 10)        dfs: list = []        for s in symbols:            # Check if file exists            file_path: Path = self.data_path.joinpath(f"{s}.csv")            if not file_path.exists():                logger.error(f"File {file_path} does not exist")                continue            # Open file            df: pl.DataFrame = pl.read_csv(file_path, schema_overrides={'date': pl.Utf8})            # Convert the date column to Date type            df = df.with_columns(                pl.col("date").str.strptime(pl.Date, "%Y%m%d").alias("date")            )            # Filter by date range            df = df.filter((pl.col("date") >= start) & (pl.col("date") <= end))            # Specify data types            df = df.with_columns(                pl.col("open"),  # .round(3).cast(pl.Float32),                pl.col("high"),  # .cast(pl.Float32),                pl.col("low"),  # .cast(pl.Float32),                pl.col("close"),  # .cast(pl.Float32),                pl.col("volume"),  # .cast(pl.Float32),            )            # Check for empty data            if df.is_empty():                continue            dfs.append(df)        #start_dates = [df.get_column("date").min() for df in dfs]        # 找到最近的起始日期        #latest_start_date = max(start_dates)        #print("统一截取的起始日期:", latest_start_date)        # dfs_cut = [        #     df.filter(pl.col("date") >= latest_start_date)        #     for df in dfs        # ]        # Concatenate results        result_df: pl.DataFrame = pl.concat(dfs)        # compare_symbol_dates(result_df,'513500.SH','512890.SH')        return result_df    def calc_exprs(self, df, names, fields):        from dataset.utility import calculate_by_expression        if len(fields) == len(names) and len(names) > 0:            results = []            for name,field in zip(names, fields):                if field == '':continue                #try:                expr = calculate_by_expression(df, field)                #except:                    # print(f'因子表达式有问题，忽略：{field}')                    # import traceback                    # traceback.print_stack()                    # continue                results.append(expr['data'].alias(name))            df = df.with_columns(results)        return df            # self.df = self.df.drop_nans()if __name__ == '__main__':    from strategy import BacktestingEngine    symbols = ['510300.SH','159915.SZ']    lab = AlphaLab(lab_path='./run', data_path='D:\work\.aitrader_data\quotes_etf')    df = lab.load_bar_df(symbols=symbols, start='20100101', end='20250506',extended_days=10)    print(df)    dataset: AlphaDataset = Alpha158(        df,        train_period=("2010-01-01", "2014-12-31"),        valid_period=("2015-01-01", "2016-12-31"),        test_period=("2017-01-01", "2020-8-31"),    )    # 添加数据预处理器    dataset.add_processor("learn", partial(process_drop_na, names=["label"]))    dataset.add_processor("learn", partial(process_cs_norm, names=["label"], method="zscore"))    # 准备特征和标签数据    name = 'etf轮动'    dataset.prepare_data(filters=None, max_workers=3)    lab.save_dataset(name=name,dataset=dataset)    dataset: AlphaDataset = lab.load_dataset(name)    model: AlphaModel = LgbModel(seed=42)    # # 使用数据集训练模型    model.fit(dataset)    # # 查看模型细节    # model.detail()    lab.save_model(name,model)    model = lab.load_model(name)    #用模型在测试集上预测    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["date", "symbol", "signal"]    dataset.show_signal_performance(signal)    lab.save_signal(name,signal)    # 从文件加载信号数据    signal = lab.load_signal(name)    import bt    from bt.algos import *    from bt_algos_extend import SelectTopK    # create the strategy    s = bt.Strategy('s1', [bt.algos.RunDaily(),                           SelectTopK(signal=signal),                           bt.algos.WeighEqually(),                           bt.algos.Rebalance()])    # 创建回测引擎对象    engine = BacktestingEngine(lab)    # 设置回测参数    engine.set_parameters(        vt_symbols=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)    # 执行回测任务    # engine.load_data()    # engine.run_backtesting()    # engine.calculate_result()    # engine.calculate_statistics()    # engine.show_chart()    #dataset.show_feature_performance("rsv_5")

典型的机器学习流程，数据及因子计算，划分数据集，然后有监督学习建模，预测信号，回测，可视化。

通过滚动式的学习，模型是可以持续进化的，这就是核心逻辑：

代码与数据下载：AI量化实验室——2025量化投资的星辰大海

吾日三省吾身

《拿铁因素》里三句话：先投资自己，让账户自动运转，从现在起，富有地生活。

其实“拿铁因素”本身，只是有效“省钱”的方式。

让账户自动运转才是根本。

很幸运这几年构建了自己的投资理财体系，账户可以自动运转。

无论你是通过“节约拿铁因素”去节流，还是通过“一人企业”去开源，这个“杠杆”的金钱系统都有有效地发挥作用。

“从现在起，富有地生活”。——不留遗憾。

如果仅仅指望退休生活，那就是《百万富翁快车道》里看不上的“慢车道”。

“有花堪折直须折，莫待无花空折枝”。

当你三十岁的时候,去买十八岁时想买的东西，去二十岁想去的地方，已经毫无意义。世界上没有那么多的来日方长,只有世事无常，欲买桂花同载酒，终不似少年游。”

如何平衡现在和未来呢？既要活在当下，又要长期主义。

其实也特别简单，存下每天收入的1/8以上（先投资自己），然后让复利滚动起来（让账户自动运转）。然后就可以在“当下”，“富有地生活”了。

不必想象自己达到A7.7或者A8，或者等到这一天。

因为你等到这一天，你会发现，其实没有特别根本的变化。

前两句话是如何实现财富自由，而第三句话则是关于为什么这么做？

财务自由并不是目的，财富自由是为了精神自由，能自由，深度去体验这个世界。首先确定对你来说什么是重要的，然后照着去做。比如你想上很久的摄影课、想看很久的演唱会、想去很久的一场旅行……你比你想象的更富有，所以从现在起富有地生活吧，不是在遥远的未来，就从今天开始。

——像你已经财富自由了那样，去生活。

大家都各大有各的事情，但多数人终归是普通人。

有人说，潜在的机会在弱联系里，这是有一定的道理。

弱联系的主动维护和激活： 弱连接不会自动产生价值。你需要：

- 建立连接： 积极参与社交活动、线上社区、行业会议等。
- 保持可见度： 偶尔互动（如社交媒体点赞评论、节日问候、分享有用信息）。
- 清晰表达价值： 在适当的时候，让对方知道你的专长和需求（但不要过于功利）。
- 乐于助人： 在力所能及的范围内帮助弱连接对象，建立好感与互惠基础。
- 善于倾听和提问： 了解对方的世界，发现潜在的交集。
  
  “潜在”二字很关键：并非所有弱连接都会带来直接机会，但弱连接网络大大增加了你接触到潜在机会的可能性。它是一个概率游戏，网络越大越多元，可能性越高。