Py学习  »  Python

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

七年实现财富自由 • 2 月前 • 134 次点击  
原创内容第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)
有了数据了因子表达式,就可以进行策略开发和回测了。
代码下载地址:AI量化实验室——量化投资的星辰大海
吾日三省吾身
回过头看,年轻时,什么都没有,学了一点本领,一穷二白出道。
没有人指点,没有人领路,好像也没有通过阅读去获取认知。
但就是那么自信。
总觉得一切都会有,而且很快就会有。
那时候想象中的世界,是比尔盖茨的创富神话。
其实在学校就遇到了很好的人,毕业后进入了很好的行业,很好的公司。
只是年少轻狂,觉得一切都还没开始呢。
什么家庭,房子,婚姻没有概念。
这就是普通家庭孩子要走的弯路吧。
“七年一辈子”,就这样年少轻狂了“一辈子”,转眼而立之年。
家里开始催婚,那时赶上房价飞涨。
开始面对现实,从梦中醒来。
创业、职业发展,买房成家生娃,又是“一辈子”。
务实,脚踏实地的一辈子,过上了世俗意义上中产生活,令爸妈满意的生活。
只有自己知道,内心的“不甘”与“挣扎”,外部不确定的风云变幻。
第三个“一辈子”,潜龙勿用,稳扎稳打。
构建ABCZ体系:人生计划之"ABCZ"
ABZ个人是比较满意的,都朝着令人满意的方向发展。
C计划,上次聊过:以交易为生:普通人快速跨越阶层的可能路径之一
这是一条路,不过不一定适合所有人。
擅长,喜欢有价值,缺一不可。
如何投资里的分散标准,以及人生之无限游戏。
专业的投资能力,可以是之一,但不建议是唯一。
毕竟,这条路一样充满荆棘。
代码和数据下载:AI量化实验室——2025量化投资的星辰大海

AI量化实验室 星球,已经运行三年多,1700+会员。

aitrader代码,因子表达式引擎、遗传算法(Deap)因子挖掘引等,支持vnpy,qlib,backtrader和bt引擎,内置多个年化30%+的策略,每周五迭代一次,代码和数据在星球全部开源。

点击 “查看原文”,直接访问策略集合

扩展  •  历史文章   

EarnMore(赚得更多)基于RL的投资组合管理框架:一致的股票表示,可定制股票池管理。(附论文+代码)

机器学习驱动的策略开发通过流程 | 普通人阶层跃迁的可能路径?

年化30.24%,最大回撤19%,综合动量多因子评分策略再升级(python代码+数据)

三秒钟创建一个年化28%,夏普比1.25的策略(python系统已开放源代码下载)

6年年化收益46%,最大回撤率为16%的策略(附python代码

Python社区是高质量的Python/Django开发社区
本文地址:http://www.python88.com/topic/185712