简介:这份德意志银行2024年策略研究报告《China Equity Strategy: Trend change, not just a rally》由特许金融分析师Peter Milliken撰写,面向关注中国股市与全球资产配置的投资者、研究员及金融从业者,帮助读者理解中国股权市场的趋势变化与潜在机会。报告围绕港股连续四年下跌、沪深300三年表现不佳的背景,探讨GDP增长与市场表现的相关性、货币政策与资金注入的作用、中国6.5万亿美元现金储备的释放前景,以及股市相对银行存款的吸引力,并指出中国市场在全球组合中被低估的现状。资源包内含1个PDF文件,大小约1.05MB,内容为完整研报原文,包含关键信息摘要、市场历史数据统计表格及分析师认证与利益披露附录,便于直接查阅与引用。目前已有89人学习下载,适合需要从宏观策略视角把握中国权益市场趋势、进行投资决策参考的读者。
1. 从一份 2024 年德意志银行中国股票策略研报说起
2024 年有一份德意志银行的中国股票策略研报,标题里带了一个很关键的词:Trend change。做二级市场投研的人看到这个词,第一反应通常不是"看多还是看空",而是"趋势结构是不是变了"。这份报告讨论的不是某只个股的买卖点,而是中国股票资产在配置框架里的定位是否发生了阶段性切换。对量化研究员、策略分析师、以及需要把卖方研报转成可执行信号的人来说,这类报告的价值不在于结论,而在于它给出的分析维度和数据口径。
问题在于,卖方研报是 PDF,是给人读的,不是给系统读的。一份几十页的策略报告里,真正能落到模型里的可能只有几张图、几个估值分位、几段资金流描述。如果每次都靠人工摘录,覆盖几十家投行的研报就崩了。所以这篇要讲的是:拿到一份像"China Equity Strategy Trend change"这样的策略研报,怎么把它拆成结构化数据,怎么提取趋势判断的关键字段,怎么用代码复现它的核心逻辑,以及在这个过程中哪些参数和口径最容易踩坑。
适合的读者是:做投研数据工程的、写策略回测的、以及需要批量处理卖方 PDF 的分析师。不需要你是德银的客户,也不需要你有原始报告,下面讲的方法对同类策略研报通用。
2. 策略研报 PDF 的结构拆解与字段设计
2.1 为什么策略研报比财报更难解析
财报有相对固定的三张表,XBRL 一上,字段就齐了。策略研报完全不是这个路子。它的正文是叙述性的,图表是嵌入的,关键数字散落在句子中间,比如"当前 MSCI China 前瞻市盈率约为 9.5 倍,低于过去十年均值一个标准差"。这句话里至少有四个可提取字段:指标名、当前值、历史均值、偏离幅度。但每个分析师的行文习惯不同,有的写"9.5x",有的写"9.5 倍",有的把数字放在括号里。
所以策略研报的解析目标不是"还原全文",而是"抽取趋势判断相关的结构化字段"。我一般会把字段分成三类:估值类(PE、PB、ERP、股息率)、资金类(南向资金、外资持仓、成交额占比)、情绪与配置类(仓位分位、超配低配、盈利预期修正)。这三类基本覆盖了"Trend change"这种标题想表达的东西。
2.2 用 pdfplumber 抽取文本与表格的最小代码
先解决"把 PDF 变成可处理文本"这一步。常见做法是用 pdfplumber,它对表格和文字位置的保留比 PyPDF2 好。
import pdfplumber import re def extract_pdf(path): pages = [] with pdfplumber.open(path) as pdf: for i, page in enumerate(pdf.pages): text = page.extract_text() or "" tables = page.extract_tables() pages.append({ "page": i + 1, "text": text, "tables": tables }) return pages def find_valuation_sentences(text): # 匹配包含估值指标和数字的句子 pattern = re.compile( r"[^。.]*?(市盈率|PE|PB|市净率|ERP|股息率)[^。.]*?\d+\.?\d*\s*(倍|x|%)[^。.]*" ) return pattern.findall(text) pages = extract_pdf("strategy_report.pdf") for p in pages[:5]: hits = find_valuation_sentences(p["text"]) if hits: print(p["page"], hits)这段代码的逻辑是:逐页抽取文本和表格,然后用正则去捞包含估值关键词和数字的句子。extract_text()对纯文字页够用,extract_tables()用来兜底那些把估值分位做成表格的页面。正则里的[^。.]*?是非贪婪匹配,避免一句话跨太多内容。实际用的时候,中英文句号都要考虑,因为这类报告经常中英混排。
参数上要注意两点:一是pdfplumber.open对扫描版 PDF 无效,如果报告是图片型的,得先走 OCR;二是extract_tables在跨页表格上会断,需要按表头做二次拼接。
2.3 趋势判断字段的口径统一表
抽出来只是第一步,口径不统一,后面没法比较。下面这张表是我处理这类策略研报时会先定好的字段映射。
| 原始表述 | 统一字段名 | 单位 | 常见口径陷阱 |
|---|---|---|---|
| 前瞻市盈率 / forward PE | fwd_pe | 倍 | 是否用未来 12 个月盈利 |
| 股权风险溢价 / ERP | erp | % | 无风险利率取 10Y 还是 1Y |
| 南向资金净流入 | southbound_flow | 亿元 | 是否含 ETF 申赎 |
| 外资持仓占比 | foreign_holding_pct | % | 自由流通市值还是总市值 |
| 盈利预期修正 | eps_revision | % | 上调家数减下调家数 |
口径不统一是策略研报解析里最大的坑。同一份报告里,PE 可能一处用 MSCI China,一处用沪深 300,如果不记录指数口径,回测时就会把两个不同标的的估值混在一起。我的做法是在字段名后面强制带指数后缀,比如fwd_pe_msci_china,宁可字段名长一点。
3. 把 Trend change 转成可计算信号的实现路径
3.1 趋势切换的三种可量化定义
"Trend change"是个模糊词,落到代码里必须先定义清楚。我一般会用三种可计算的定义,分别对应不同的时间尺度。
第一种是估值分位切换:当前估值在过去 N 年分位从低于 30% 升到高于 50%,视为估值趋势切换。第二种是资金流方向切换:南向资金连续 M 周净流入转净流出,或反之。第三种是盈利预期修正方向切换:EPS 修正动量由负转正。
import pandas as pd def valuation_regime_switch(pe_series, window=252*5, low=0.3, high=0.5): # 计算滚动分位 rank = pe_series.rolling(window).apply( lambda x: pd.Series(x).rank(pct=True).iloc[-1], raw=False ) # 从低位区间进入中位以上,标记为切换 signal = ((rank.shift(1) < low) & (rank > high)).astype(int) return rank, signal def flow_direction_switch(flow_series, m=4): # 滚动 m 期求和,判断符号变化 roll = flow_series.rolling(m).sum() sign = (roll > 0).astype(int) switch = (sign.diff() != 0).astype(int) return roll, switchvaluation_regime_switch里window控制分位回溯长度,默认取 5 年交易日;low和high是切换的阈值带,设成 0.3 和 0.5 是为了避免在 0.49 到 0.51 之间反复触发。flow_direction_switch里的m是平滑窗口,用 4 周是为了过滤单周噪声。这两个函数输出的 signal 都是 0/1 序列,可以直接接到回测框架里。
3.2 用 pandas 复现研报里的估值分位图
策略研报里最常见的图就是估值分位带。要复现它,核心是滚动分位计算。注意rolling().apply()在大样本上很慢,实务里我会用numpy的sliding_window_view加速。
import numpy as np def rolling_percentile_fast(arr, window): from numpy.lib.stride_tricks import sliding_window_view windows = sliding_window_view(arr, window) # 对每个窗口算最后一个值在窗口内的分位 pct = (windows < windows[:, -1][:, None]).sum(axis=1) / window out = np.full(len(arr), np.nan) out[window-1:] = pct return outsliding_window_view不复制数据,只改视图,所以内存占用低。windows[:, -1]取每个窗口的最后一个值,(windows < ...).sum(axis=1) / window就是分位。这个实现比rolling().apply()快一个数量级,适合处理十年以上的日频数据。参数window要和研报里声明的回溯期一致,德银这类报告常用 10 年,但图注里有时写的是"since 2010",要按实际起点算。
3.3 信号回测的最小框架与参数
有了信号,下一步是验证它有没有用。最小回测框架只需要价格序列和信号序列。
def backtest(price, signal, cost=0.001): ret = price.pct_change().fillna(0) # 信号滞后一期,避免未来函数 pos = signal.shift(1).fillna(0) strat_ret = pos * ret - pos.diff().abs().fillna(0) * cost cum = (1 + strat_ret).cumprod() return cum, strat_ret cum, sr = backtest(price_series, signal_series, cost=0.0015)signal.shift(1)是关键,研报里的判断通常是基于当期数据得出的,实盘只能下一期执行,不滞后就是未来函数。cost默认 0.1%,跨境策略要调高到 0.15% 到 0.2%,因为涉及汇率和交易摩擦。输出的cum是净值曲线,sr是逐期收益,可以进一步算夏普和最大回撤。
注意:研报里的回测往往不含交易成本,直接照搬会高估收益。自己复现时至少要把换手成本加进去。
4. 批量处理多份策略研报的工程化与排错
4.1 多报告字段对齐的合并逻辑
单份报告解析通了,真正的需求是批量。几十份研报来自不同投行,字段名、单位、指数口径都不一样。合并时我一般分两步:先按统一字段名做列映射,再按日期和指数口径做行对齐。
FIELD_MAP = { "forward PE": "fwd_pe", "前瞻市盈率": "fwd_pe", "P/E (x)": "fwd_pe", "ERP": "erp", "股权风险溢价": "erp", } def normalize(df, field_map): df = df.rename(columns=field_map) # 只保留统一字段 keep = [c for c in df.columns if c in set(field_map.values())] return df[keep] def merge_reports(dfs, keys=("date", "index_name")): from functools import reduce merged = reduce(lambda a, b: pd.merge(a, b, on=list(keys), how="outer"), dfs) return merged.sort_values("date")FIELD_MAP是人工维护的,这是没办法省的一步,因为自然语言表述太发散。normalize只保留映射后的统一字段,避免不同报告的冗余列污染合并结果。merge_reports用 outer join,是因为不同报告覆盖的日期和指数不完全重合,inner join 会丢数据。合并后一定要检查缺失率,某个字段缺失超过 50% 就要考虑是不是映射错了。
4.2 解析失败的常见原因与排查顺序
批量跑的时候,失败是常态。我一般按这个顺序排查:
- 先看是不是扫描版 PDF,用
page.extract_text()返回空字符串判断。 - 再看编码问题,中文报告有时用 CID 字体,抽出来是乱码,需要换
pdfminer的LAParams参数。 - 然后看表格跨页,
extract_tables在跨页处会返回 None,需要按表头补。 - 最后看正则匹配率,如果某份报告匹配到的句子数远低于均值,多半是行文风格差异,要单独加规则。
def diagnose(path): with pdfplumber.open(path) as pdf: p0 = pdf.pages[0] text = p0.extract_text() or "" if len(text.strip()) < 50: return "可能是扫描版,需要 OCR" if "�" in text: return "编码异常,检查字体嵌入" return "文本层正常"这个诊断函数只做粗筛,但能快速把问题报告分堆。实务里我会把失败报告单独存一个目录,人工过一遍再决定是加规则还是放弃。
4.3 用配置表管理不同投行的解析规则
与其把规则写死在代码里,不如抽成配置。下面是一个 YAML 配置的示例结构。
db_2024_china: bank: "Deutsche Bank" year: 2024 index_scope: ["MSCI China", "CSI 300"] valuation_pattern: "(前瞻市盈率|forward PE|P/E)" flow_pattern: "(南向资金|southbound)" lookback_years: 10 cost_bps: 15index_scope限定这份报告涉及的指数,避免把不同标的的估值混算。lookback_years和cost_bps直接喂给前面的分位计算和回测函数。这样新增一家投行的报告,只加一段配置,不动主逻辑。配置化之后,批量处理的维护成本会明显下降。
5. 从研报信号到组合权重的进阶处理
5.1 把离散信号转成连续权重
前面输出的 signal 是 0/1,实盘里更常见的是连续权重。一个简单做法是用分位本身做权重,而不是等权。
def signal_to_weight(rank, signal, cap=1.0): # 只在信号触发时给权重,权重与分位偏离成正比 raw = np.where(signal == 1, (rank - 0.5) * 2, 0) raw = np.clip(raw, 0, cap) return raw / raw.sum() if raw.sum() > 0 else raw(rank - 0.5) * 2把分位从 [0.5, 1] 映射到 [0, 1],np.clip限制单资产上限,最后归一化。这样估值分位越高,权重越大,但不会无限集中。cap参数控制单资产最大权重,跨境组合一般设 0.4 到 0.5。
5.2 信号衰减与再平衡频率的选择
研报里的趋势判断通常不是高频信号,再平衡频率设太高会被交易成本吃掉。我一般会对比周度、双周、月度三种频率下的净值曲线。
| 再平衡频率 | 年换手率(估) | 适用场景 |
|---|---|---|
| 周度 | 15-25 倍 | 资金流驱动信号 |
| 双周 | 8-15 倍 | 估值分位信号 |
| 月度 | 4-8 倍 | 配置型策略 |
估值分位这类慢变量,月度再平衡通常就够了。资金流信号变化快,可以到周度,但要把cost_bps相应调高。判断标准不是哪个收益高,而是扣费后夏普是否还稳定。
5.3 用滚动窗口检验信号的稳定性
最后一个技巧是滚动窗口检验。把样本切成多个三年窗口,分别跑回测,看信号在不同区间是否都有效。
def rolling_backtest(price, signal, window=252*3, step=252): results = [] for start in range(0, len(price) - window, step): seg_p = price.iloc[start:start+window] seg_s = signal.iloc[start:start+window] cum, sr = backtest(seg_p, seg_s) results.append({ "start": seg_p.index[0], "end": seg_p.index[-1], "sharpe": sr.mean() / sr.std() * (252 ** 0.5) if sr.std() > 0 else 0 }) return pd.DataFrame(results)window取三年是为了让每个子区间有足够样本,step取一年保证窗口之间有重叠但不完全重复。输出的夏普序列如果波动很大,说明信号不稳定,可能只在特定市场环境下有效。这一步比看全样本净值更能暴露问题,也是把一份策略研报真正吃透的收尾动作。
本文还有配套的精品资源,点击获取