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

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")