简介本资源是一套面向金融从业者、量化初学者及高校财经/计算机专业学生的Python金融实务与量化分析系统课程覆盖从编程基础到实战建模的完整能力链路。内容包含16个章节的PDF讲义系统讲解Python基础语法、NumPy/Pandas金融数据处理、Matplotlib可视化、网络爬虫开发含多场景实战、金融函数应用、信用评分卡构建、可转债定价模型等核心模块并配套2019年系列教程与量化因子、基本面投资专题课件。资源为RAR压缩包共117.67MB内含数十份结构清晰的教学文档类型以PDF为主涵盖理论讲解、代码示例与实操推导便于分阶段学习与反复研读。目前已有571人学习下载适合希望夯实Python金融编程能力、掌握量化分析底层逻辑并获取可复用知识框架的学习者。1. 这不是又一套“Python入门课”而是一份能直接跑通A股因子回测、信用评分卡建模、可转债定价的金融实务工作流你手头可能有十几份标着“Python金融”的PDF但打开后发现全是print(Hello World)和for循环求斐波那契——这种内容离真实交易台或风控系统差了至少三道防火墙。这份课程包不同它用16个章节串联起一条完整链路——从用pandas.read_csv()加载沪深300成分股日线数据到调用scipy.optimize.minimize()求解最小方差组合权重从用requestsBeautifulSoup抓取巨潮网公告PDF链接到用sklearn.linear_model.LogisticRegression训练违约概率模型甚至包含可转债的BSM修正模型与二叉树定价实现。它不教你怎么装Python而是默认你已配好conda环境直接切入pandas.DataFrame.resample(M).last()处理月度财务指标、statsmodels.tsa.adfuller()检验因子平稳性、plotly.graph_objects.Figure.add_trace()渲染多因子IC值热力图。适合两类人刚转行进券商量化部/银行风控岗的新人需要快速产出可汇报的分析脚本以及已有2年Python经验但没碰过真实金融数据结构的开发者急需补上“如何把财报附注里的非结构化文本转成可计算字段”这一环。2. 用pandasnumpy构建金融数据管道从原始CSV到因子矩阵的标准化处理2.1 为什么必须重写read_csv参数金融数据的5类隐性陷阱金融数据源如聚宽、Tushare、本地Excel导出常埋着5类pandas.read_csv()默认行为无法处理的坑日期列自动转换失败2023-03-15被识别为字符串而非datetime64导致df[trade_date].dt.month报错千分位逗号干扰数值解析1,234.56被读作字符串astype(float)抛出ValueError空值标记不统一NULL、N/A、-、空白单元格混用na_values未指定则全变为NaN列名含不可见字符Excel导出时列名末尾带\xa0不间断空格df[close]报KeyError编码错误导致中文列名乱码GB2312编码文件用UTF-8读取列名变b\xc4\xda\xb9\xc9。课程中Python课程2.1.pdf给出的生产级读取模板如下import pandas as pd import numpy as np def load_financial_csv(filepath, date_coltrade_date, numeric_colsNone): 加载金融CSV的鲁棒函数 :param filepath: 文件路径 :param date_col: 日期列名支持多列如[trade_date,report_date] :param numeric_cols: 需强制转数值的列名列表如[open,high,low,close,volume] # 步骤1预读取前10行探测编码避免gbk/utf-8误判 with open(filepath, rb) as f: raw f.read(1000) encoding gbk if b\x81\x40 in raw else utf-8 # 步骤2指定所有潜在空值标记 na_values [NULL, N/A, -, , , None, nan] # 步骤3读取并清洗列名 df pd.read_csv( filepath, encodingencoding, na_valuesna_values, skipinitialspaceTrue, # 忽略列名前后空格 keep_default_naFalse # 禁用pandas默认na值完全由na_values控制 ) # 步骤4清洗列名去除不可见字符、空格、特殊符号 df.columns df.columns.str.replace(r[^\w\s], , regexTrue).str.strip() df.columns df.columns.str.replace(r\s, _, regexTrue) # 多空格→单下划线 # 步骤5转换日期列 if isinstance(date_col, str): date_col [date_col] for col in date_col: if col in df.columns: df[col] pd.to_datetime(df[col], errorscoerce) # 错误值转NaT # 步骤6转换数值列处理千分位逗号 if numeric_cols: for col in numeric_cols: if col in df.columns: # 先转字符串再移除逗号再转float df[col] df[col].astype(str).str.replace(,, ).astype(float) return df # 实际调用示例加载某券商提供的A股日线数据 stock_daily load_financial_csv( a_share_daily_2023.csv, date_coltrade_date, numeric_cols[open, high, low, close, volume, amount] ) print(f加载成功{len(stock_daily)} 行列名{list(stock_daily.columns)})提示errorscoerce在pd.to_datetime()中至关重要——它将无法解析的日期如2023-02-30转为NaTNot a Time而非中断整个流程。这是金融数据清洗的第一道安全阀。2.2 构建因子矩阵用pandas的groupbyapply实现跨股票时间序列标准化量化策略的核心输入是因子矩阵rows股票×时间cols因子值。课程量化金融基础课 -量化因子.pdf指出直接对全量数据df[pe_ratio].rank(pctTrue)会忽略“同一时间点不同股票的可比性”必须按时间切片标准化。以下是课程中推荐的zscore_by_date函数def zscore_by_date(df, value_col, group_coltrade_date, min_periods10, ddof0): 按日期分组对因子值做Z-Score标准化 :param df: 输入DataFrame需含group_col和value_col :param value_col: 待标准化的因子列名 :param group_col: 分组列名通常为日期 :param min_periods: 每组最少有效样本数低于则结果为NaN :param ddof: 自由度修正ddof0即总体标准差ddof1为样本标准差 def _zscore_group(group): # 过滤掉该组内缺失值 valid_data group[value_col].dropna() if len(valid_data) min_periods: return pd.Series([np.nan] * len(group), indexgroup.index) mean_val valid_data.mean() std_val valid_data.std(ddofddof) if std_val 0: return pd.Series([0.0] * len(group), indexgroup.index) # 对原group中每一行计算z-score保持索引对齐 result (group[value_col] - mean_val) / std_val return result # 使用transform确保返回与原df等长的Series return df.groupby(group_col, sortFalse)[value_col].transform(_zscore_group) # 应用示例对PE_TTM因子做每日横截面标准化 stock_daily[pe_zscore] zscore_by_date( stock_daily, value_colpe_ttm, group_coltrade_date, min_periods50 # 要求每日至少50只股票有有效PE值 ) print(PE_TTM Z-Score标准化完成前5行) print(stock_daily[[trade_date, symbol, pe_ttm, pe_zscore]].head())2.2.1 为什么不用scipy.stats.zscore()——内存与索引对齐的硬约束课程明确指出scipy.stats.zscore()接受一维数组若直接传入df.groupby(trade_date)[pe_ttm].apply(lambda x: zscore(x))返回的是Series其索引是日期无法直接赋值给原df。而transform()保证输出Series的索引与原df完全一致这是金融数据管道中不可妥协的特性。此外transform()内部优化了分组计算对百万级数据比循环for date, group in df.groupby(trade_date)快3倍以上。2.3 用numpy加速因子计算避免pandas apply的Python层开销当因子逻辑涉及复杂数学运算如滚动夏普比率、最大回撤pandas.DataFrame.apply()的Python循环成为瓶颈。课程Python课程2.1.pdf演示了如何用numpy.lib.stride_tricks.sliding_window_view替代import numpy as np from numpy.lib.stride_tricks import sliding_window_view def rolling_max_drawdown(returns, window252): 计算滚动窗口内的最大回撤基于收益率序列 :param returns: 一维numpy array日收益率 :param window: 滚动窗口长度交易日 :return: 与returns等长的array每位置为截至该日的window期内最大回撤 # 步骤1计算累计净值假设初始净值为1 cumprod np.cumprod(1 returns) # 步骤2生成滑动窗口视图每行是一个window长度的子数组 windows sliding_window_view(cumprod, window_shapewindow) # 步骤3对每个窗口计算最大回撤 1 - min(窗口内净值)/max(窗口内净值的前缀最大值) # 注意需对每个窗口单独计算前缀最大值 drawdowns np.zeros(len(returns)) for i in range(len(windows)): win windows[i] # 前缀最大值cummax[0]win[0], cummax[1]max(win[0],win[1]), ... cummax np.maximum.accumulate(win) # 当前回撤 1 - 当前净值 / 历史最高净值 dd 1 - win / cummax drawdowns[i window - 1] np.max(dd) # 最大回撤赋给窗口末尾位置 return drawdowns # 对某只股票的收益率序列应用假设returns_series是pandas Series returns_np stock_daily[stock_daily[symbol]000001.SZ][pct_chg].values / 100.0 mdd_array rolling_max_drawdown(returns_np, window252) # 将结果映射回原df需对齐索引 stock_daily.loc[stock_daily[symbol]000001.SZ, mdd_252] mdd_array注意sliding_window_view在numpy 1.20才引入。若环境为旧版课程提供备选方案——用numba.jit编译加速循环代码见Python课程1.5.pdf第7页。3. 从爬虫到因子用requestsBeautifulSoup抓取并结构化财报关键字段3.1 爬虫环境配置的关键避坑点User-Agent、Session复用与反爬响应处理课程Python课程3.1.pdf强调金融数据爬虫失败的80%原因不在代码逻辑而在HTTP请求头配置。巨潮网、上交所等平台对无User-Agent或Accept-Language的请求直接返回403。更隐蔽的是部分页面要求Cookie中存在JSESSIONID而该值需通过首次GET首页获取。课程给出的标准Session初始化模板import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def create_financial_session(): 创建专用于金融网站爬取的Session对象 session requests.Session() # 设置通用headers模拟主流浏览器 session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0.0.0 Safari/537.36, Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/avif,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, Connection: keep-alive, Upgrade-Insecure-Requests: 1, Sec-Fetch-Dest: document, Sec-Fetch-Mode: navigate, Sec-Fetch-Site: none, Sec-Fetch-User: ?1, }) # 配置重试策略网络抖动常见 retry_strategy Retry( total3, backoff_factor1, status_forcelist[429, 500, 502, 503, 504], ) adapter HTTPAdapter(max_retriesretry_strategy) session.mount(http://, adapter) session.mount(https://, adapter) # 关键访问首页以获取必要Cookie如巨潮网的JSESSIONID try: session.get(http://www.cninfo.com.cn/, timeout10) except Exception as e: print(f首页预热失败但继续执行{e}) return session # 实例化全局复用避免重复创建Session fin_session create_financial_session()3.2 解析年报PDF中的关键财务指标从文本提取到结构化存储课程python课程4.1-Python金融函数与金融建模.pdf指出直接OCR PDF效率低且错误率高。更优路径是先用爬虫定位年报PDF下载链接再用PyPDF2或pdfplumber提取文本最后用正则匹配关键字段。以下为提取“归属于母公司股东的净利润”年报中常见表述的实战代码import pdfplumber import re import pandas as pd def extract_net_profit_from_pdf(pdf_path): 从PDF年报中提取“归属于母公司股东的净利润”数值 :param pdf_path: PDF文件路径 :return: float单位万元若未找到则返回np.nan net_profit np.nan # 使用pdfplumber比PyPDF2更擅长表格和文本定位 with pdfplumber.open(pdf_path) as pdf: # 优先扫描“合并利润表”所在页通常在P15-P30 for page_num in range(min(30, len(pdf)), 15, -1): try: page pdf.pages[page_num] text page.extract_text() if not text: continue # 匹配多种表述年报用语不统一 patterns [ r归属于母公司股东的净利润.*?([\d,]\.?\d*)\s*(?:万元|元), r净利润\(归属于母公司股东\).*?([\d,]\.?\d*)\s*(?:万元|元), r归属于母公司所有者的净利润.*?([\d,]\.?\d*)\s*(?:万元|元), r净利润.*?归属于母公司股东.*?([\d,]\.?\d*)\s*(?:万元|元), ] for pattern in patterns: match re.search(pattern, text, re.DOTALL | re.IGNORECASE) if match: # 提取数字并转为float处理千分位逗号 num_str match.group(1).replace(,, ) net_profit float(num_str) # 若原文单位为“元”转为“万元” if 元 in match.group(0): net_profit / 10000.0 break if not np.isnan(net_profit): break except Exception as e: continue # 跳过解析失败的页面 return net_profit # 批量处理示例假设已爬取PDF列表 pdf_files [000001_2022_annual_report.pdf, 000002_2022_annual_report.pdf] results [] for pdf in pdf_files: profit extract_net_profit_from_pdf(pdf) results.append({symbol: pdf[:6], net_profit_wan: profit}) profit_df pd.DataFrame(results) print(提取结果) print(profit_df)3.2.1 为什么不用PDF表格提取——年报中“净利润”常不在标准表格内课程特别说明年报的“合并利润表”虽为表格但“归属于母公司股东的净利润”常作为表格最后一行加粗显示或在“管理层讨论与分析”章节以文字形式重申。pdfplumber的extract_text()能捕获这些非表格区域而extract_table()会遗漏。这是金融文档解析中必须绕过的典型陷阱。4. 信用评分卡建模全流程从WOE分箱到Logistic回归验证4.1 WOE分箱的工业级实现用sklearn的KBinsDiscretizer规避手动cut陷阱课程python课程4.2-信用评分卡.pdf指出新手常犯错误是用pandas.cut()对连续变量分箱导致边界值重复、区间不连续。课程推荐使用sklearn.preprocessing.KBinsDiscretizer并封装为woe_binning函数from sklearn.preprocessing import KBinsDiscretizer from sklearn.base import BaseEstimator, TransformerMixin import numpy as np import pandas as pd class WOEBinner(BaseEstimator, TransformerMixin): WOE分箱器对特征进行等频/等宽分箱并计算各箱WOE值 def __init__(self, n_bins10, strategyquantile, encodeordinal): self.n_bins n_bins self.strategy strategy self.encode encode self.binner_ None self.woe_map_ {} def fit(self, X, y): X: 特征矩阵pandas DataFrame或numpy array y: 目标变量0/11表示违约 # 初始化分箱器 self.binner_ KBinsDiscretizer( n_binsself.n_bins, strategyself.strategy, encodeself.encode ) # 对每列特征单独拟合避免多列同时分箱的干扰 for col in X.columns: # 只对数值列操作 if np.issubdtype(X[col].dtype, np.number): # 拟合分箱器注意需reshape为2D X_col X[col].values.reshape(-1, 1) self.binner_.fit(X_col) # 获取分箱后的标签0,1,2,... bins self.binner_.transform(X_col).toarray() if self.encodeonehot else self.binner_.transform(X_col) bin_labels bins.flatten().astype(int) if self.encodeordinal else np.argmax(bins, axis1) # 计算WOEWOE ln( (bad_i / total_bad) / (good_i / total_good) ) total_bad np.sum(y 1) total_good np.sum(y 0) woe_dict {} for bin_id in np.unique(bin_labels): mask bin_labels bin_id bad_in_bin np.sum(y[mask] 1) good_in_bin np.sum(y[mask] 0) # 平滑处理避免0除 p_bad (bad_in_bin 0.5) / (total_bad 1.0) p_good (good_in_bin 0.5) / (total_good 1.0) woe_dict[bin_id] np.log(p_bad / p_good) if p_good 0 else 0.0 self.woe_map_[col] woe_dict return self def transform(self, X): 将X转换为WOE编码 X_woe X.copy() for col in X.columns: if col in self.woe_map_: # 重新分箱获取bin_id X_col X[col].values.reshape(-1, 1) bin_labels self.binner_.transform(X_col).flatten().astype(int) # 映射为WOE值 X_woe[col] [self.woe_map_[col].get(b, 0.0) for b in bin_labels] return X_woe # 使用示例假设已有特征矩阵X_train和目标y_train # woe_binner WOEBinner(n_bins5, strategyquantile) # X_train_woe woe_binner.fit_transform(X_train, y_train) # X_test_woe woe_binner.transform(X_test)4.2 评分卡逻辑回归的KS值验证与分数映射课程强调建模结束不等于完成。必须验证区分能力KS值并将logit分数映射为业务可读的“信用分”。以下是课程中calculate_ks和score_mapping函数from sklearn.metrics import roc_curve import numpy as np def calculate_ks(y_true, y_score): 计算KS值Kolmogorov-Smirnov statistic :param y_true: 真实标签0/1 :param y_score: 模型预测概率0~1 :return: KS值0~1 fpr, tpr, _ roc_curve(y_true, y_score) ks np.max(np.abs(tpr - fpr)) return ks def score_mapping(y_score, base_score600, pdo50, odds1/19): 将logistic回归概率映射为整数信用分 :param y_score: 模型输出的概率0~1 :param base_score: 基准分当odds1:19时的分数 :param pdo: 好坏比翻倍所需的分数增量 :param odds: 基准好坏比好客户数/坏客户数 # 将概率转logitlogit ln(p/(1-p)) logit np.log(y_score / (1 - y_score 1e-8)) # 防止除零 # 分数公式Score A - B * logit # 其中 B PDO / ln(2), A BaseScore B * ln(odds) B pdo / np.log(2) A base_score B * np.log(odds) scores A - B * logit return np.round(scores).astype(int) # 假设已训练好逻辑回归模型 # from sklearn.linear_model import LogisticRegression # lr LogisticRegression() # lr.fit(X_train_woe, y_train) # y_pred_proba lr.predict_proba(X_test_woe)[:, 1] # ks_value calculate_ks(y_test, y_pred_proba) # credit_scores score_mapping(y_pred_proba) # print(fKS值{ks_value:.3f}0.3为优秀) # print(f信用分范围{credit_scores.min()} ~ {credit_scores.max()})提示score_mapping中的odds1/19意味着“每19个好客户对应1个坏客户”这是银行业常用基准。若实际样本好坏比偏离此值需调整odds参数使分数分布合理。5. 可转债定价实战用二叉树模型实现含回售条款的估值5.1 二叉树节点构建考虑转股价下修与回售权的动态调整课程python课程4.3-可转债定价.pdf指出标准二叉树Cox-Ross-Rubinstein无法处理可转债特有的条款。课程实现的ConvertibleBondTree类在每个节点动态判断是否触发下修或回售import numpy as np class ConvertibleBondTree: 可转债二叉树定价器支持转股价下修、回售条款 def __init__(self, S0, K, r, sigma, T, N, put_price0, put_date0, down_adj_ratio0.8, down_adj_freq1): :param S0: 标的股票初始价格 :param K: 初始转股价 :param r: 无风险利率 :param sigma: 股票波动率 :param T: 到期时间年 :param N: 二叉树步数 :param put_price: 回售价格如103元 :param put_date: 回售起始时间年如1.5年 :param down_adj_ratio: 转股价下修比例如0.8表示下修20% :param down_adj_freq: 下修频率年如1年可下修一次 self.S0 S0 self.K K self.r r self.sigma sigma self.T T self.N N self.put_price put_price self.put_date put_date self.down_adj_ratio down_adj_ratio self.down_adj_freq down_adj_freq self.dt T / N self.u np.exp(sigma * np.sqrt(self.dt)) self.d 1 / self.u self.p (np.exp(r * self.dt) - self.d) / (self.u - self.d) def price(self, coupon_rate0.02, face_value100): 计算可转债理论价格 # 初始化股价网格S[i][j]表示第i步第j个节点的股价 S np.zeros((self.N1, self.N1)) for i in range(self.N1): for j in range(i1): S[i][j] self.S0 * (self.u ** (i-j)) * (self.d ** j) # 初始化转股价网格K_grid[i][j]表示第i步第j个节点的转股价 K_grid np.full_like(S, self.K) # 动态更新转股价当股价下跌超阈值且满足下修条件时下调 for i in range(1, self.N1): for j in range(i1): # 计算当前时间点年 t i * self.dt # 检查是否满足下修条件股价低于原转股价*80%且距上次下修满1年 if S[i][j] self.K * self.down_adj_ratio and t self.down_adj_freq: # 查找上一次下修时间简化假设首次下修即永久生效 K_grid[i][j] self.K * self.down_adj_ratio else: # 继承父节点转股价上一步的两个父节点中取较小者因下修不可逆 if i 1: parent_up K_grid[i-1][j-1] if j 0 else self.K parent_down K_grid[i-1][j] if j i else self.K K_grid[i][j] min(parent_up, parent_down) # 初始化期权价值网格V[i][j]表示第i步第j个节点的债券价值 V np.zeros((self.N1, self.N1)) # 终端节点按转股价值与面值利息较大者取值但不超过回售价 for j in range(self.N1): # 转股价值 max(0, S[N][j] - K_grid[N][j]) * 100 / K_grid[N][j] 假设100元面值转1股 conversion_value max(0, S[self.N][j] - K_grid[self.N][j]) * face_value / K_grid[self.N][j] # 债券价值 面值 累计利息 bond_value face_value * (1 coupon_rate * self.T) # 终端价值 max(转股价值, 债券价值) V[self.N][j] max(conversion_value, bond_value) # 回售条款若当前时间回售起始时间且转股价值回售价则可回售 t_now self.N * self.dt if t_now self.put_date and conversion_value self.put_price: V[self.N][j] self.put_price # 倒推计算从倒数第二步开始 for i in range(self.N-1, -1, -1): for j in range(i1): # 计算子节点价值的期望现值 up_val V[i1][j] down_val V[i1][j1] expected_val self.p * up_val (1 - self.p) * down_val discounted_val np.exp(-self.r * self.dt) * expected_val # 当前节点价值 max(转股价值, 折现后子节点价值, 回售价若满足条件) conversion_value_here max(0, S[i][j] - K_grid[i][j]) * face_value / K_grid[i][j] current_val max(conversion_value_here, discounted_val) t_now i * self.dt if t_now self.put_date and conversion_value_here self.put_price: current_val max(current_val, self.put_price) V[i][j] current_val return V[0][0] # 实例化并定价示例参数 cb_tree ConvertibleBondTree( S010.0, # 正股当前价10元 K12.0, # 初始转股价12元 r0.02, # 无风险利率2% sigma0.3, # 波动率30% T3.0, # 3年期 N100, # 100步二叉树 put_price103.0, # 回售价103元 put_date1.5, # 1.5年后可回售 down_adj_ratio0.8, # 可下修至80% down_adj_freq1.0 # 每1年可下修 ) price cb_tree.price(coupon_rate0.015, face_value100) print(f可转债理论价格{price:.2f} 元)5.1.1 为什么必须动态更新转股价——条款博弈的真实体现课程解释转股价下修不是一次性事件而是发行人根据股价走势、市场情绪、监管窗口期动态决策的过程。二叉树中每个节点代表一个可能的未来情景必须在该情景下判断“是否满足下修条件”而非简单地在某个固定时间点下调。这正是该模型区别于教科书式二叉树的核心。6. 量化因子回测的陷阱排查从数据泄露到幸存者偏差的5个检查清单6.1 因子回测前必做的5项数据完整性验证课程量化金融基础课 -股票量化基本面投资.pdf将回测失败归因于数据问题的比例定为73%。以下是课程推荐的factor_backtest_sanity_check函数覆盖最致命的5类漏洞import pandas as pd import numpy as np def factor_backtest_sanity_check(factor_df, price_df, date_coltrade_date, symbol_colsymbol, factor_colfactor_value): 对因子数据进行5项核心完整性检查 :param factor_df: 因子DataFrame含date_col, symbol_col, factor_col :param price_df: 价格DataFrame含date_col, symbol_col, close等 :return: dict检查结果摘要 checks {} # 检查1日期范围一致性因子数据不能超前于价格数据 factor_dates set(factor_df[date_col].unique()) price_dates set(price_df[date_col].unique()) future_dates factor_dates - price_dates checks[date_leakage] len(future_dates) 0 if not checks[date_leakage]: checks[leaked_dates] sorted(list(future_dates))[:5] # 只显示前5个 # 检查2股票池覆盖度因子覆盖的股票数 vs 价格数据中的股票数 factor_symbols set(factor_df[symbol_col].unique()) price_symbols set(price_df[symbol_col].unique()) uncovered_symbols price_symbols - factor_symbols checks[symbol_coverage] len(uncovered_symbols) / len(price_symbols) 0.1 # 要求覆盖90% checks[uncovered_count] len(uncovered_symbols) # 检查3因子值分布异常极端值、全零、全NaN factor_series factor_df[factor_col].dropna() checks[factor_distribution] { n_total: len(factor_df), n_valid: len(factor_series), valid_ratio: len(factor_series) / len(factor_df), min: factor_series.min(), max: factor_series.max(), std: factor_series.std(), n_extreme_outliers: len(factor_series[np.abs(factor_series - factor_series.mean()) 5 * factor_series.std()]) } # 检查4时间序列连续性检查是否存在“跳空”日期 sorted_dates sorted(factor_df[date_col].unique()) date_diffs np.diff(sorted_dates) checks[date_gaps] np.any(date_diffs np.timedelta64(10, D)) # 超过10天视为断点 # 检查5幸存者偏差检测检查因子是否仅存在于当前上市股票 # 方法对比因子数据中最早的日期与股票 p a hrefhttps://download.csdn.net/download/qq_27595745/82478037 stylecolor:#ec7500;font-size:14px; 本文还有配套的精品资源点击获取 /a img altmenu-r.4af5f7ec.gif srchttps://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif stylewidth:16px;margin-left:4px;vertical-align:text-bottom;cursor:text; /p
金融Python实战：A股因子回测、信用评分卡与可转债定价
简介本资源是一套面向金融从业者、量化初学者及高校财经/计算机专业学生的Python金融实务与量化分析系统课程覆盖从编程基础到实战建模的完整能力链路。内容包含16个章节的PDF讲义系统讲解Python基础语法、NumPy/Pandas金融数据处理、Matplotlib可视化、网络爬虫开发含多场景实战、金融函数应用、信用评分卡构建、可转债定价模型等核心模块并配套2019年系列教程与量化因子、基本面投资专题课件。资源为RAR压缩包共117.67MB内含数十份结构清晰的教学文档类型以PDF为主涵盖理论讲解、代码示例与实操推导便于分阶段学习与反复研读。目前已有571人学习下载适合希望夯实Python金融编程能力、掌握量化分析底层逻辑并获取可复用知识框架的学习者。1. 这不是又一套“Python入门课”而是一份能直接跑通A股因子回测、信用评分卡建模、可转债定价的金融实务工作流你手头可能有十几份标着“Python金融”的PDF但打开后发现全是print(Hello World)和for循环求斐波那契——这种内容离真实交易台或风控系统差了至少三道防火墙。这份课程包不同它用16个章节串联起一条完整链路——从用pandas.read_csv()加载沪深300成分股日线数据到调用scipy.optimize.minimize()求解最小方差组合权重从用requestsBeautifulSoup抓取巨潮网公告PDF链接到用sklearn.linear_model.LogisticRegression训练违约概率模型甚至包含可转债的BSM修正模型与二叉树定价实现。它不教你怎么装Python而是默认你已配好conda环境直接切入pandas.DataFrame.resample(M).last()处理月度财务指标、statsmodels.tsa.adfuller()检验因子平稳性、plotly.graph_objects.Figure.add_trace()渲染多因子IC值热力图。适合两类人刚转行进券商量化部/银行风控岗的新人需要快速产出可汇报的分析脚本以及已有2年Python经验但没碰过真实金融数据结构的开发者急需补上“如何把财报附注里的非结构化文本转成可计算字段”这一环。2. 用pandasnumpy构建金融数据管道从原始CSV到因子矩阵的标准化处理2.1 为什么必须重写read_csv参数金融数据的5类隐性陷阱金融数据源如聚宽、Tushare、本地Excel导出常埋着5类pandas.read_csv()默认行为无法处理的坑日期列自动转换失败2023-03-15被识别为字符串而非datetime64导致df[trade_date].dt.month报错千分位逗号干扰数值解析1,234.56被读作字符串astype(float)抛出ValueError空值标记不统一NULL、N/A、-、空白单元格混用na_values未指定则全变为NaN列名含不可见字符Excel导出时列名末尾带\xa0不间断空格df[close]报KeyError编码错误导致中文列名乱码GB2312编码文件用UTF-8读取列名变b\xc4\xda\xb9\xc9。课程中Python课程2.1.pdf给出的生产级读取模板如下import pandas as pd import numpy as np def load_financial_csv(filepath, date_coltrade_date, numeric_colsNone): 加载金融CSV的鲁棒函数 :param filepath: 文件路径 :param date_col: 日期列名支持多列如[trade_date,report_date] :param numeric_cols: 需强制转数值的列名列表如[open,high,low,close,volume] # 步骤1预读取前10行探测编码避免gbk/utf-8误判 with open(filepath, rb) as f: raw f.read(1000) encoding gbk if b\x81\x40 in raw else utf-8 # 步骤2指定所有潜在空值标记 na_values [NULL, N/A, -, , , None, nan] # 步骤3读取并清洗列名 df pd.read_csv( filepath, encodingencoding, na_valuesna_values, skipinitialspaceTrue, # 忽略列名前后空格 keep_default_naFalse # 禁用pandas默认na值完全由na_values控制 ) # 步骤4清洗列名去除不可见字符、空格、特殊符号 df.columns df.columns.str.replace(r[^\w\s], , regexTrue).str.strip() df.columns df.columns.str.replace(r\s, _, regexTrue) # 多空格→单下划线 # 步骤5转换日期列 if isinstance(date_col, str): date_col [date_col] for col in date_col: if col in df.columns: df[col] pd.to_datetime(df[col], errorscoerce) # 错误值转NaT # 步骤6转换数值列处理千分位逗号 if numeric_cols: for col in numeric_cols: if col in df.columns: # 先转字符串再移除逗号再转float df[col] df[col].astype(str).str.replace(,, ).astype(float) return df # 实际调用示例加载某券商提供的A股日线数据 stock_daily load_financial_csv( a_share_daily_2023.csv, date_coltrade_date, numeric_cols[open, high, low, close, volume, amount] ) print(f加载成功{len(stock_daily)} 行列名{list(stock_daily.columns)})提示errorscoerce在pd.to_datetime()中至关重要——它将无法解析的日期如2023-02-30转为NaTNot a Time而非中断整个流程。这是金融数据清洗的第一道安全阀。2.2 构建因子矩阵用pandas的groupbyapply实现跨股票时间序列标准化量化策略的核心输入是因子矩阵rows股票×时间cols因子值。课程量化金融基础课 -量化因子.pdf指出直接对全量数据df[pe_ratio].rank(pctTrue)会忽略“同一时间点不同股票的可比性”必须按时间切片标准化。以下是课程中推荐的zscore_by_date函数def zscore_by_date(df, value_col, group_coltrade_date, min_periods10, ddof0): 按日期分组对因子值做Z-Score标准化 :param df: 输入DataFrame需含group_col和value_col :param value_col: 待标准化的因子列名 :param group_col: 分组列名通常为日期 :param min_periods: 每组最少有效样本数低于则结果为NaN :param ddof: 自由度修正ddof0即总体标准差ddof1为样本标准差 def _zscore_group(group): # 过滤掉该组内缺失值 valid_data group[value_col].dropna() if len(valid_data) min_periods: return pd.Series([np.nan] * len(group), indexgroup.index) mean_val valid_data.mean() std_val valid_data.std(ddofddof) if std_val 0: return pd.Series([0.0] * len(group), indexgroup.index) # 对原group中每一行计算z-score保持索引对齐 result (group[value_col] - mean_val) / std_val return result # 使用transform确保返回与原df等长的Series return df.groupby(group_col, sortFalse)[value_col].transform(_zscore_group) # 应用示例对PE_TTM因子做每日横截面标准化 stock_daily[pe_zscore] zscore_by_date( stock_daily, value_colpe_ttm, group_coltrade_date, min_periods50 # 要求每日至少50只股票有有效PE值 ) print(PE_TTM Z-Score标准化完成前5行) print(stock_daily[[trade_date, symbol, pe_ttm, pe_zscore]].head())2.2.1 为什么不用scipy.stats.zscore()——内存与索引对齐的硬约束课程明确指出scipy.stats.zscore()接受一维数组若直接传入df.groupby(trade_date)[pe_ttm].apply(lambda x: zscore(x))返回的是Series其索引是日期无法直接赋值给原df。而transform()保证输出Series的索引与原df完全一致这是金融数据管道中不可妥协的特性。此外transform()内部优化了分组计算对百万级数据比循环for date, group in df.groupby(trade_date)快3倍以上。2.3 用numpy加速因子计算避免pandas apply的Python层开销当因子逻辑涉及复杂数学运算如滚动夏普比率、最大回撤pandas.DataFrame.apply()的Python循环成为瓶颈。课程Python课程2.1.pdf演示了如何用numpy.lib.stride_tricks.sliding_window_view替代import numpy as np from numpy.lib.stride_tricks import sliding_window_view def rolling_max_drawdown(returns, window252): 计算滚动窗口内的最大回撤基于收益率序列 :param returns: 一维numpy array日收益率 :param window: 滚动窗口长度交易日 :return: 与returns等长的array每位置为截至该日的window期内最大回撤 # 步骤1计算累计净值假设初始净值为1 cumprod np.cumprod(1 returns) # 步骤2生成滑动窗口视图每行是一个window长度的子数组 windows sliding_window_view(cumprod, window_shapewindow) # 步骤3对每个窗口计算最大回撤 1 - min(窗口内净值)/max(窗口内净值的前缀最大值) # 注意需对每个窗口单独计算前缀最大值 drawdowns np.zeros(len(returns)) for i in range(len(windows)): win windows[i] # 前缀最大值cummax[0]win[0], cummax[1]max(win[0],win[1]), ... cummax np.maximum.accumulate(win) # 当前回撤 1 - 当前净值 / 历史最高净值 dd 1 - win / cummax drawdowns[i window - 1] np.max(dd) # 最大回撤赋给窗口末尾位置 return drawdowns # 对某只股票的收益率序列应用假设returns_series是pandas Series returns_np stock_daily[stock_daily[symbol]000001.SZ][pct_chg].values / 100.0 mdd_array rolling_max_drawdown(returns_np, window252) # 将结果映射回原df需对齐索引 stock_daily.loc[stock_daily[symbol]000001.SZ, mdd_252] mdd_array注意sliding_window_view在numpy 1.20才引入。若环境为旧版课程提供备选方案——用numba.jit编译加速循环代码见Python课程1.5.pdf第7页。3. 从爬虫到因子用requestsBeautifulSoup抓取并结构化财报关键字段3.1 爬虫环境配置的关键避坑点User-Agent、Session复用与反爬响应处理课程Python课程3.1.pdf强调金融数据爬虫失败的80%原因不在代码逻辑而在HTTP请求头配置。巨潮网、上交所等平台对无User-Agent或Accept-Language的请求直接返回403。更隐蔽的是部分页面要求Cookie中存在JSESSIONID而该值需通过首次GET首页获取。课程给出的标准Session初始化模板import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def create_financial_session(): 创建专用于金融网站爬取的Session对象 session requests.Session() # 设置通用headers模拟主流浏览器 session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0.0.0 Safari/537.36, Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/avif,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, Connection: keep-alive, Upgrade-Insecure-Requests: 1, Sec-Fetch-Dest: document, Sec-Fetch-Mode: navigate, Sec-Fetch-Site: none, Sec-Fetch-User: ?1, }) # 配置重试策略网络抖动常见 retry_strategy Retry( total3, backoff_factor1, status_forcelist[429, 500, 502, 503, 504], ) adapter HTTPAdapter(max_retriesretry_strategy) session.mount(http://, adapter) session.mount(https://, adapter) # 关键访问首页以获取必要Cookie如巨潮网的JSESSIONID try: session.get(http://www.cninfo.com.cn/, timeout10) except Exception as e: print(f首页预热失败但继续执行{e}) return session # 实例化全局复用避免重复创建Session fin_session create_financial_session()3.2 解析年报PDF中的关键财务指标从文本提取到结构化存储课程python课程4.1-Python金融函数与金融建模.pdf指出直接OCR PDF效率低且错误率高。更优路径是先用爬虫定位年报PDF下载链接再用PyPDF2或pdfplumber提取文本最后用正则匹配关键字段。以下为提取“归属于母公司股东的净利润”年报中常见表述的实战代码import pdfplumber import re import pandas as pd def extract_net_profit_from_pdf(pdf_path): 从PDF年报中提取“归属于母公司股东的净利润”数值 :param pdf_path: PDF文件路径 :return: float单位万元若未找到则返回np.nan net_profit np.nan # 使用pdfplumber比PyPDF2更擅长表格和文本定位 with pdfplumber.open(pdf_path) as pdf: # 优先扫描“合并利润表”所在页通常在P15-P30 for page_num in range(min(30, len(pdf)), 15, -1): try: page pdf.pages[page_num] text page.extract_text() if not text: continue # 匹配多种表述年报用语不统一 patterns [ r归属于母公司股东的净利润.*?([\d,]\.?\d*)\s*(?:万元|元), r净利润\(归属于母公司股东\).*?([\d,]\.?\d*)\s*(?:万元|元), r归属于母公司所有者的净利润.*?([\d,]\.?\d*)\s*(?:万元|元), r净利润.*?归属于母公司股东.*?([\d,]\.?\d*)\s*(?:万元|元), ] for pattern in patterns: match re.search(pattern, text, re.DOTALL | re.IGNORECASE) if match: # 提取数字并转为float处理千分位逗号 num_str match.group(1).replace(,, ) net_profit float(num_str) # 若原文单位为“元”转为“万元” if 元 in match.group(0): net_profit / 10000.0 break if not np.isnan(net_profit): break except Exception as e: continue # 跳过解析失败的页面 return net_profit # 批量处理示例假设已爬取PDF列表 pdf_files [000001_2022_annual_report.pdf, 000002_2022_annual_report.pdf] results [] for pdf in pdf_files: profit extract_net_profit_from_pdf(pdf) results.append({symbol: pdf[:6], net_profit_wan: profit}) profit_df pd.DataFrame(results) print(提取结果) print(profit_df)3.2.1 为什么不用PDF表格提取——年报中“净利润”常不在标准表格内课程特别说明年报的“合并利润表”虽为表格但“归属于母公司股东的净利润”常作为表格最后一行加粗显示或在“管理层讨论与分析”章节以文字形式重申。pdfplumber的extract_text()能捕获这些非表格区域而extract_table()会遗漏。这是金融文档解析中必须绕过的典型陷阱。4. 信用评分卡建模全流程从WOE分箱到Logistic回归验证4.1 WOE分箱的工业级实现用sklearn的KBinsDiscretizer规避手动cut陷阱课程python课程4.2-信用评分卡.pdf指出新手常犯错误是用pandas.cut()对连续变量分箱导致边界值重复、区间不连续。课程推荐使用sklearn.preprocessing.KBinsDiscretizer并封装为woe_binning函数from sklearn.preprocessing import KBinsDiscretizer from sklearn.base import BaseEstimator, TransformerMixin import numpy as np import pandas as pd class WOEBinner(BaseEstimator, TransformerMixin): WOE分箱器对特征进行等频/等宽分箱并计算各箱WOE值 def __init__(self, n_bins10, strategyquantile, encodeordinal): self.n_bins n_bins self.strategy strategy self.encode encode self.binner_ None self.woe_map_ {} def fit(self, X, y): X: 特征矩阵pandas DataFrame或numpy array y: 目标变量0/11表示违约 # 初始化分箱器 self.binner_ KBinsDiscretizer( n_binsself.n_bins, strategyself.strategy, encodeself.encode ) # 对每列特征单独拟合避免多列同时分箱的干扰 for col in X.columns: # 只对数值列操作 if np.issubdtype(X[col].dtype, np.number): # 拟合分箱器注意需reshape为2D X_col X[col].values.reshape(-1, 1) self.binner_.fit(X_col) # 获取分箱后的标签0,1,2,... bins self.binner_.transform(X_col).toarray() if self.encodeonehot else self.binner_.transform(X_col) bin_labels bins.flatten().astype(int) if self.encodeordinal else np.argmax(bins, axis1) # 计算WOEWOE ln( (bad_i / total_bad) / (good_i / total_good) ) total_bad np.sum(y 1) total_good np.sum(y 0) woe_dict {} for bin_id in np.unique(bin_labels): mask bin_labels bin_id bad_in_bin np.sum(y[mask] 1) good_in_bin np.sum(y[mask] 0) # 平滑处理避免0除 p_bad (bad_in_bin 0.5) / (total_bad 1.0) p_good (good_in_bin 0.5) / (total_good 1.0) woe_dict[bin_id] np.log(p_bad / p_good) if p_good 0 else 0.0 self.woe_map_[col] woe_dict return self def transform(self, X): 将X转换为WOE编码 X_woe X.copy() for col in X.columns: if col in self.woe_map_: # 重新分箱获取bin_id X_col X[col].values.reshape(-1, 1) bin_labels self.binner_.transform(X_col).flatten().astype(int) # 映射为WOE值 X_woe[col] [self.woe_map_[col].get(b, 0.0) for b in bin_labels] return X_woe # 使用示例假设已有特征矩阵X_train和目标y_train # woe_binner WOEBinner(n_bins5, strategyquantile) # X_train_woe woe_binner.fit_transform(X_train, y_train) # X_test_woe woe_binner.transform(X_test)4.2 评分卡逻辑回归的KS值验证与分数映射课程强调建模结束不等于完成。必须验证区分能力KS值并将logit分数映射为业务可读的“信用分”。以下是课程中calculate_ks和score_mapping函数from sklearn.metrics import roc_curve import numpy as np def calculate_ks(y_true, y_score): 计算KS值Kolmogorov-Smirnov statistic :param y_true: 真实标签0/1 :param y_score: 模型预测概率0~1 :return: KS值0~1 fpr, tpr, _ roc_curve(y_true, y_score) ks np.max(np.abs(tpr - fpr)) return ks def score_mapping(y_score, base_score600, pdo50, odds1/19): 将logistic回归概率映射为整数信用分 :param y_score: 模型输出的概率0~1 :param base_score: 基准分当odds1:19时的分数 :param pdo: 好坏比翻倍所需的分数增量 :param odds: 基准好坏比好客户数/坏客户数 # 将概率转logitlogit ln(p/(1-p)) logit np.log(y_score / (1 - y_score 1e-8)) # 防止除零 # 分数公式Score A - B * logit # 其中 B PDO / ln(2), A BaseScore B * ln(odds) B pdo / np.log(2) A base_score B * np.log(odds) scores A - B * logit return np.round(scores).astype(int) # 假设已训练好逻辑回归模型 # from sklearn.linear_model import LogisticRegression # lr LogisticRegression() # lr.fit(X_train_woe, y_train) # y_pred_proba lr.predict_proba(X_test_woe)[:, 1] # ks_value calculate_ks(y_test, y_pred_proba) # credit_scores score_mapping(y_pred_proba) # print(fKS值{ks_value:.3f}0.3为优秀) # print(f信用分范围{credit_scores.min()} ~ {credit_scores.max()})提示score_mapping中的odds1/19意味着“每19个好客户对应1个坏客户”这是银行业常用基准。若实际样本好坏比偏离此值需调整odds参数使分数分布合理。5. 可转债定价实战用二叉树模型实现含回售条款的估值5.1 二叉树节点构建考虑转股价下修与回售权的动态调整课程python课程4.3-可转债定价.pdf指出标准二叉树Cox-Ross-Rubinstein无法处理可转债特有的条款。课程实现的ConvertibleBondTree类在每个节点动态判断是否触发下修或回售import numpy as np class ConvertibleBondTree: 可转债二叉树定价器支持转股价下修、回售条款 def __init__(self, S0, K, r, sigma, T, N, put_price0, put_date0, down_adj_ratio0.8, down_adj_freq1): :param S0: 标的股票初始价格 :param K: 初始转股价 :param r: 无风险利率 :param sigma: 股票波动率 :param T: 到期时间年 :param N: 二叉树步数 :param put_price: 回售价格如103元 :param put_date: 回售起始时间年如1.5年 :param down_adj_ratio: 转股价下修比例如0.8表示下修20% :param down_adj_freq: 下修频率年如1年可下修一次 self.S0 S0 self.K K self.r r self.sigma sigma self.T T self.N N self.put_price put_price self.put_date put_date self.down_adj_ratio down_adj_ratio self.down_adj_freq down_adj_freq self.dt T / N self.u np.exp(sigma * np.sqrt(self.dt)) self.d 1 / self.u self.p (np.exp(r * self.dt) - self.d) / (self.u - self.d) def price(self, coupon_rate0.02, face_value100): 计算可转债理论价格 # 初始化股价网格S[i][j]表示第i步第j个节点的股价 S np.zeros((self.N1, self.N1)) for i in range(self.N1): for j in range(i1): S[i][j] self.S0 * (self.u ** (i-j)) * (self.d ** j) # 初始化转股价网格K_grid[i][j]表示第i步第j个节点的转股价 K_grid np.full_like(S, self.K) # 动态更新转股价当股价下跌超阈值且满足下修条件时下调 for i in range(1, self.N1): for j in range(i1): # 计算当前时间点年 t i * self.dt # 检查是否满足下修条件股价低于原转股价*80%且距上次下修满1年 if S[i][j] self.K * self.down_adj_ratio and t self.down_adj_freq: # 查找上一次下修时间简化假设首次下修即永久生效 K_grid[i][j] self.K * self.down_adj_ratio else: # 继承父节点转股价上一步的两个父节点中取较小者因下修不可逆 if i 1: parent_up K_grid[i-1][j-1] if j 0 else self.K parent_down K_grid[i-1][j] if j i else self.K K_grid[i][j] min(parent_up, parent_down) # 初始化期权价值网格V[i][j]表示第i步第j个节点的债券价值 V np.zeros((self.N1, self.N1)) # 终端节点按转股价值与面值利息较大者取值但不超过回售价 for j in range(self.N1): # 转股价值 max(0, S[N][j] - K_grid[N][j]) * 100 / K_grid[N][j] 假设100元面值转1股 conversion_value max(0, S[self.N][j] - K_grid[self.N][j]) * face_value / K_grid[self.N][j] # 债券价值 面值 累计利息 bond_value face_value * (1 coupon_rate * self.T) # 终端价值 max(转股价值, 债券价值) V[self.N][j] max(conversion_value, bond_value) # 回售条款若当前时间回售起始时间且转股价值回售价则可回售 t_now self.N * self.dt if t_now self.put_date and conversion_value self.put_price: V[self.N][j] self.put_price # 倒推计算从倒数第二步开始 for i in range(self.N-1, -1, -1): for j in range(i1): # 计算子节点价值的期望现值 up_val V[i1][j] down_val V[i1][j1] expected_val self.p * up_val (1 - self.p) * down_val discounted_val np.exp(-self.r * self.dt) * expected_val # 当前节点价值 max(转股价值, 折现后子节点价值, 回售价若满足条件) conversion_value_here max(0, S[i][j] - K_grid[i][j]) * face_value / K_grid[i][j] current_val max(conversion_value_here, discounted_val) t_now i * self.dt if t_now self.put_date and conversion_value_here self.put_price: current_val max(current_val, self.put_price) V[i][j] current_val return V[0][0] # 实例化并定价示例参数 cb_tree ConvertibleBondTree( S010.0, # 正股当前价10元 K12.0, # 初始转股价12元 r0.02, # 无风险利率2% sigma0.3, # 波动率30% T3.0, # 3年期 N100, # 100步二叉树 put_price103.0, # 回售价103元 put_date1.5, # 1.5年后可回售 down_adj_ratio0.8, # 可下修至80% down_adj_freq1.0 # 每1年可下修 ) price cb_tree.price(coupon_rate0.015, face_value100) print(f可转债理论价格{price:.2f} 元)5.1.1 为什么必须动态更新转股价——条款博弈的真实体现课程解释转股价下修不是一次性事件而是发行人根据股价走势、市场情绪、监管窗口期动态决策的过程。二叉树中每个节点代表一个可能的未来情景必须在该情景下判断“是否满足下修条件”而非简单地在某个固定时间点下调。这正是该模型区别于教科书式二叉树的核心。6. 量化因子回测的陷阱排查从数据泄露到幸存者偏差的5个检查清单6.1 因子回测前必做的5项数据完整性验证课程量化金融基础课 -股票量化基本面投资.pdf将回测失败归因于数据问题的比例定为73%。以下是课程推荐的factor_backtest_sanity_check函数覆盖最致命的5类漏洞import pandas as pd import numpy as np def factor_backtest_sanity_check(factor_df, price_df, date_coltrade_date, symbol_colsymbol, factor_colfactor_value): 对因子数据进行5项核心完整性检查 :param factor_df: 因子DataFrame含date_col, symbol_col, factor_col :param price_df: 价格DataFrame含date_col, symbol_col, close等 :return: dict检查结果摘要 checks {} # 检查1日期范围一致性因子数据不能超前于价格数据 factor_dates set(factor_df[date_col].unique()) price_dates set(price_df[date_col].unique()) future_dates factor_dates - price_dates checks[date_leakage] len(future_dates) 0 if not checks[date_leakage]: checks[leaked_dates] sorted(list(future_dates))[:5] # 只显示前5个 # 检查2股票池覆盖度因子覆盖的股票数 vs 价格数据中的股票数 factor_symbols set(factor_df[symbol_col].unique()) price_symbols set(price_df[symbol_col].unique()) uncovered_symbols price_symbols - factor_symbols checks[symbol_coverage] len(uncovered_symbols) / len(price_symbols) 0.1 # 要求覆盖90% checks[uncovered_count] len(uncovered_symbols) # 检查3因子值分布异常极端值、全零、全NaN factor_series factor_df[factor_col].dropna() checks[factor_distribution] { n_total: len(factor_df), n_valid: len(factor_series), valid_ratio: len(factor_series) / len(factor_df), min: factor_series.min(), max: factor_series.max(), std: factor_series.std(), n_extreme_outliers: len(factor_series[np.abs(factor_series - factor_series.mean()) 5 * factor_series.std()]) } # 检查4时间序列连续性检查是否存在“跳空”日期 sorted_dates sorted(factor_df[date_col].unique()) date_diffs np.diff(sorted_dates) checks[date_gaps] np.any(date_diffs np.timedelta64(10, D)) # 超过10天视为断点 # 检查5幸存者偏差检测检查因子是否仅存在于当前上市股票 # 方法对比因子数据中最早的日期与股票 p a hrefhttps://download.csdn.net/download/qq_27595745/82478037 stylecolor:#ec7500;font-size:14px; 本文还有配套的精品资源点击获取 /a img altmenu-r.4af5f7ec.gif srchttps://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif stylewidth:16px;margin-left:4px;vertical-align:text-bottom;cursor:text; /p