基于polars的因子表达式引擎，附python代码 | 股票策略引擎开发中

原创内容第974篇，专注AGI+，AI量化投资、个人成长与财富自由。

我们将扩展全量股票市场回测的技术。

难度在于几千支股票价量数据，数十个因子甚至更多的数据加载和处理的计算量。

回测是需要加载全部数据的。

每天都去查数据库，回测效率会很低。

主要计算：

1、加载价量，财务数据（季频，日频）可以预处理成日频，并保存成hdf5。

2、基于基础数据的因子计算（因子表达式）。

从本地读取相应的行（日期范围）列数据（因子表达式所需基础因子）。

可以使用duckdb来加载，转成dataframe。而后使用因子表达式引擎计算。

aitrader教学版本规划：

1，多因子策略方向（平台化）：因子表达式引擎，回测引擎，进行etf/股票策略的回测。

平台化，尽量按低代码，可视化的方式，给不熟悉代码，不想编程的同学一个量化投资的工具。以及提供相应的策略可以学习。

2、因子挖掘、机器学习等方向的技术探索。

2，实盘方向：tqsdk + backtrader（以交易为生）。

而新技术探索及量化系统代码，是以交付源代码的方式，给有编程能力的同学提供。这时候gui就不是最重要的，以notebook演示技术为主。

统一使用PolarLoader来加载csv数据：

from pathlib import Pathimport pandas as pdimport polars as plfrom loguru import loggerfrom datetime import datetime
from alpha.dataset.utility import DataProxyfrom config import DATA_DIR

def compare_symbol_dates(df, symbol1: str, symbol2: str):    # 提取两个 Symbol 的数据    df1 = df.filter(pl.col("symbol") == symbol1)    df2 = df.filter(pl.col("symbol") == symbol2)
    # 提取唯一日期    dates1 = df1.select(pl.col("date").unique()).to_series()    dates2 = df2.select(pl.col("date").unique()).to_series()
    # 计算差异日期    diff1 = dates1.filter(~dates1.is_in(dates2))    diff2 = dates2.filter(~dates2.is_in(dates1))
    # 打印结果    if not diff1.is_empty():        print(f"=== {symbol1} 独有的日期数据 ===")        print(df1.filter(pl.col("date").is_in(diff1)))    else:        print(f"{symbol1} 和 {symbol2} 的日期完全一致")
    if not diff2.is_empty():        print(f"\n=== {symbol2} 独有的日期数据 ===")        print(df2.filter(pl.col("date").is_in(diff2)))    else:        print(f"{symbol2} 和 {symbol1} 的日期完全一致")

class PolarDataloader:    def __init__(self, symbols:list[str],                 start:str = '20100101',                 end: str = datetime.now().strftime('%Y%m%d'),                 fields=[],                 names=[],                 path:Path=DATA_DIR.joinpath('quotes')):        self.path = path        self.symbols = symbols        self.start = start        self.end = end        self.df = self._load_df_from_csvs()        if len


    
(fields) == len(names) and len(names) > 0:            results = []            for name,field in zip(names, fields):                if field == '':continue                expr = self.calculate_by_expression(self.df, field)                results.append(expr['data'].alias(name))            self.df = self.df.with_columns(results)            self.df = self.df.drop_nans()
    def _load_df_from_csvs(self, symbols=None):        if not symbols:            symbols = self.symbols
        if not self.end:            self.end = datetime.now().strftime("%Y%m%d")        dfs: list = []        for s in symbols:            # Check if file exists            file_path: Path = self.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})            df = df.sort("date")  # 升序排列            #print(df)
            # Filter by date range            df = df.filter((pl.col("date") >= self.start) & (pl.col("date") <= self.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_cut)        # # '.SH',  # 红利低波（股票）        # '.SZ',  # 豆粕（商品）        # compare_symbol_dates(result_df,'513500.SH','512890.SH')        return result_df
    def calculate_by_expression(self, df: pl.DataFrame, expression: str) -> pl.DataFrame:        from alpha.dataset.ts_function import (  # noqa            ts_delay,            ts_min, ts_max,            ts_argmax, ts_argmin,            ts_rank, ts_sum,            ts_mean, ts_std,            ts_slope, ts_quantile,            ts_rsquare, ts_resi,            ts_corr,            ts_less, ts_greater,            ts_log, ts_abs        )        from alpha.dataset.ts_function import ts_delay as shift # noqa        from alpha.dataset.expr_extends import trend_score, roc, RSRS # noqa        from alpha.dataset.ts_function import ts_mean as ma  # noqa        from alpha.dataset.ts_function import ts_slope as slope # noqa
        # Extract feature objects to local space        d: dict


    
 = locals()
        for column in df.columns:            # Filter index columns            if column in {"date", "symbol"}:                continue
            # Cache feature df            column_df = df[["date", "symbol", column]]            d[column] = DataProxy(column_df)
        # Use eval to execute calculation        other: DataProxy = eval(expression, {}, d)

        return other.df


    def get_col_df(self, col='close'):        # print(self.df)        # duplicates = self.df.filter(        #     pl.count().over(['date', 'symbol']) > 1        # ).select(['date', 'symbol'])        #        # duplicated_records = self.df.join(duplicates, on=['date', 'symbol'], how='semi')        # print('recored!!!',duplicated_records)
        #        df_col = self.df.pivot(values=col, index='date', on='symbol',aggregate_function = "first").sort('date')  # 假设日期列名为 'date'
        df_col = df_col.to_pandas()        #df_col.dropna(inplace=True)        df_col.set_index('date', inplace=True)        df_col = df_col.ffill()
        #df_col.fillna(0, inplace=True)        df_col.index = pd.to_datetime(df_col.index)        # 或者更简洁的写法（直接筛选数字类型）：        #df_col = df_col.round(3)        #df_col.to_csv('close.csv')        return df_col

    def get_col_df_by_symbols(self,symbols: list[str], col='close'):        df_all = self._load_df_from_csvs(symbols=symbols)        if col not in df_all.columns:            logger.error('{}列不存在')            return None        df_close = df_all.pivot(values=col, index='date', on='symbol').sort('date')  # 假设日期列名为 'date'        # df_close = df_close.filter(        #     (pl.col('date') >= start_date) &        #     (pl.col('date') <= end_date)        # )        df_col = df_close.to_pandas()        df_col.set_index('date', inplace=True)        df_col = df_col.ffill()        df_col.index = pd.to_datetime(df_col.index)        return df_col

if __name__ == '__main__':    loader = PolarDataloader(symbols=['510300.SH','159915.SZ'],start='20200101', names=['roc_5'], fields=['close/shift(close,5)-1'])    df = loader.get_col_df('roc_5')    print(df)    df = loader.get_col_df('high')    print(df)    df = loader.get_col_df_by_symbols(symbols=['159915.SZ'],col='high')    print(df)