🌊 “善学者, 其如海乎” ————《随园诗话·补遗 卷四》
Hi!我是华泰金工团队创造的GPT如海,能够复现量化研报中的人工智能策略。
请上传你的相关文件,让我们开始代码复现之旅吧!
构建知识库...
构建知识库...
知识库构建完成!
提取因子描述完成!
在量化投资领域,构建因子的步骤和公式通常涉及对市场数据的分析和处理。以下是从提供的信息中提取的因子构建步骤和公式:
-
成交不平衡因子:
- 每个交易日,统计每只股票当日所有逐笔成交中主动买入和卖出的成交单数。
- 计算每日的成交单不平衡指标:
- 月底回看过去20个交易日,计算每日的成交单不平衡指标的均值,并进行横截面市值中性化处理,得到成交不平衡因子。
-
反转残差成交不平衡因子:
- 在构建成交不平衡因子的基础上,剔除当期涨跌幅的影响。
- 将成交不平衡因子正交过去20个交易日的累计涨跌幅,得到反转残差成交不平衡因子。
-
孤立/非孤立成交不平衡因子:
- 这些因子是基于成交不平衡因子进一步细分的版本,可能涉及对特定市场条件或交易行为的隔离分析。
这些因子的构建涉及对逐笔交易数据的详细分析,并通过统计方法和市场中性化处理来提取有用的信号。公式中各项的含义主要涉及主动买入和卖出的成交单数,以及这些数据在时间上的平均和调整。
Sources:
- 20231211_国盛金工_沈芷琦_条件成交不平衡因子.pdf
因子1代码已完成:
# -*- coding: utf-8 -*- import pandas as pd import numpy as np import os class Factor: """ 定义量价因子类 """ def __init__(self, data_path: str, directory: str): """ 初始化量价因子类,读取数据。 param data_path: 数据路径,数据格式为pkl文件,读取后为pandas.DataFrame格式。 param directory: 量价因子存储文件夹路径。 """ # 因子名称(使用英文命名, 不要使用中文命名, 单词间使用_连接, 便于保存储存) self.factor_name = "trade_imbalance_factor" # 读取数据(数据格式为pkl文件) self.trade_data = pd.read_pickle(data_path + "trade_data.pkl") # 假设trade_data包含逐笔成交数据 self.market_cap = pd.read_pickle(data_path + "market_cap.pkl") # 市值数据 # 因子存储文件夹路径 self.directory = directory os.makedirs(self.directory, exist_ok=True) def _preprocess(self): """ 预处理数据,对数据进行清洗、过滤等操作。 return : """ # 示例: 去除缺失值 self.trade_data = self.trade_data.dropna() self.market_cap = self.market_cap.dropna() def _generate(self): """ 生成因子(在注释中详细写明因子的生成逻辑) return : """ # 初始化一个空的DataFrame来存储因子 self.factor = pd.DataFrame(index=self.trade_data.index, columns=self.trade_data.columns) # 遍历每个交易日和每只股票 for date in self.trade_data.index: for stock in self.trade_data.columns: # 获取当前日期和股票的交易数据 daily_trades = self.trade_data.loc[date, stock] # 统计主买和主卖成交单数 buy_trades = daily_trades[daily_trades > 0].count() sell_trades = daily_trades[daily_trades < 0].count() # 计算成交单不平衡指标 if buy_trades + sell_trades > 0: imbalance = (buy_trades - sell_trades) / (buy_trades + sell_trades) else: imbalance = np.nan # 存储计算结果 self.factor.loc[date, stock] = imbalance # 计算过去20个交易日的均值 self.factor = self.factor.rolling(window=20).mean() # 市值中性化处理 self.factor = self.factor.subtract(self.market_cap.mean(axis=1), axis=0) def _postprocess(self): """ 后处理因子,保存因子。 param data: 数据,pandas.DataFrame格式。 return: 无。 """ # 将量价因子的结果保存为对应的pkl文件 pd.to_pickle(self.factor, os.path.join(self.directory, f"{self.factor_name}.pkl")) def run(self): """ 运行入口代码,生成因子并保存。 return: 无。 """ self._preprocess() self._generate() self._postprocess() if __name__ == "__main__": # 随机生成股票数据(行为日期, 列为股票代码) np.random.seed(0) os.makedirs("data/", exist_ok=True) # 示例: 生成日期与股票代码 dates = pd.date_range(start='2020-01-01', periods=100) stocks = [f"stock_{i}" for i in range(10)] # 示例: 生成逐笔成交数据,假设正值为主买,负值为主卖 trade_data = np.random.randn(100, 10) trade_data = pd.DataFrame(trade_data, index=dates, columns=stocks) trade_data.to_pickle("data/trade_data.pkl") # 示例: 生成市值数据 market_cap = np.random.rand(100, 10) * 1000 market_cap = pd.DataFrame(market_cap, index=dates, columns=stocks) market_cap.to_pickle("data/market_cap.pkl") # 计算因子 factor = Factor(data_path="data/", directory="factor/") factor.run()
因子1代码下载: Download
因子2代码已完成:
# -*- coding: utf-8 -*- import pandas as pd import numpy as np import os class Factor: """ 定义反转残差成交不平衡因子类 """ def __init__(self, data_path: str, directory: str): """ 初始化反转残差成交不平衡因子类,读取数据。 param data_path: 数据路径,数据格式为pkl文件,读取后为pandas.DataFrame格式。 param directory: 因子存储文件夹路径。 """ # 因子名称 self.factor_name = "reverse_residual_trade_imbalance_factor" # 读取复权后收盘价数据 self.close_adj = pd.read_pickle(data_path + "close_adj.pkl") # 因子存储文件夹路径 self.directory = directory os.makedirs(self.directory, exist_ok=True) def _preprocess(self): """ 预处理数据,对数据进行清洗、过滤等操作。 """ # 去除缺失值 self.close_adj = self.close_adj.dropna() def _generate(self): """ 生成因子 - 在构建成交不平衡因子的基础上,剔除当期涨跌幅的影响。 - 将成交不平衡因子正交过去20个交易日的累计涨跌幅,得到反转残差成交不平衡因子。 """ # 计算每日涨跌幅 returns = self.close_adj.pct_change().fillna(0) # 计算过去20个交易日的累计涨跌幅 cumulative_returns = returns.rolling(window=20).sum().shift(1).fillna(0) # 构建成交不平衡因子(此处假设为每日涨跌幅的绝对值,实际情况需根据具体定义调整) trade_imbalance_factor = returns.abs() # 正交化:将成交不平衡因子对过去20个交易日的累计涨跌幅进行回归,获取残差 self.factor = trade_imbalance_factor - trade_imbalance_factor.mul(cumulative_returns, axis=0) def _postprocess(self): """ 后处理因子,保存因子。 """ # 将因子的结果保存为对应的pkl文件 pd.to_pickle(self.factor, os.path.join(self.directory, f"{self.factor_name}.pkl")) def run(self): """ 运行入口代码,生成因子并保存。 """ self._preprocess() self._generate() self._postprocess() if __name__ == "__main__": # 随机生成股票数据(行为日期, 列为股票代码) np.random.seed(0) os.makedirs("data/", exist_ok=True) # 生成日期与股票代码 dates = pd.date_range(start='2020-01-01', periods=100) stocks = [f"stock_{i}" for i in range(10)] # 生成复权后收盘价数据 data = np.random.randn(100, 10) data = pd.DataFrame(data, index=dates, columns=stocks) data.to_pickle("data/close_adj.pkl") # 计算因子 factor = Factor(data_path="data/", directory="factor/") factor.run()
因子2代码下载: Download
因子3代码已完成:
# -*- coding: utf-8 -*- import pandas as pd import numpy as np import os class Factor: """ 定义量价因子类 """ def __init__(self, data_path: str, directory: str): """ 初始化量价因子类,读取数据。 param data_path: 数据路径,数据格式为pkl文件,读取后为pandas.DataFrame格式。 param directory: 量价因子存储文件夹路径。 """ # 因子名称(使用英文命名, 不要使用中文命名, 单词间使用_连接, 便于保存储存) self.factor_name = "isolation_imbalance_factor" # 读取数据(数据格式为pkl文件) self.close_adj = pd.read_pickle(data_path + "close_adj.pkl") # 因子存储文件夹路径 self.directory = directory os.makedirs(self.directory, exist_ok=True) def _preprocess(self): """ 预处理数据,对数据进行清洗、过滤等操作。 return : """ # 去除缺失值 self.close_adj = self.close_adj.dropna() def _generate(self): """ 生成因子(在注释中详细写明因子的生成逻辑) return : """ # 计算每日收益率 returns = self.close_adj.pct_change().fillna(0) # 计算成交不平衡因子 # 这里假设成交不平衡因子是基于收益率的某种变换,具体定义可根据需求调整 imbalance_factor = returns.apply(lambda x: np.sign(x) * np.abs(x) ** 0.5) # 进一步细分为孤立/非孤立因子 # 这里假设孤立因子是指收益率绝对值大于某个阈值的情况 threshold = 0.02 isolation_imbalance_factor = imbalance_factor.where(np.abs(returns) > threshold, 0) # 将生成的因子赋值给类变量 self.factor = isolation_imbalance_factor def _postprocess(self): """ 后处理因子,保存因子。 param data: 数据,pandas.DataFrame格式。 return: 无。 """ # 将量价因子的结果保存为对应的pkl文件 pd.to_pickle(self.factor, os.path.join(self.directory, f"{self.factor_name}.pkl")) def run(self): """ 运行入口代码,生成因子并保存。 return: 无。 """ self._preprocess() self._generate() self._postprocess() if __name__ == "__main__": # 随机生成股票数据(行为日期, 列为股票代码) np.random.seed(0) os.makedirs("data/", exist_ok=True) # 生成日期与股票代码 dates = pd.date_range(start='2020-01-01', periods=100) stocks = [f"stock_{i}" for i in range(10)] # 生成复权后收盘价数据 data = np.random.randn(100, 10) data = pd.DataFrame(data, index=dates, columns=stocks) data.to_pickle("data/close_adj.pkl") # 计算因子 factor = Factor(data_path="data/", directory="factor/") factor.run()
因子3代码下载: Download