第 8 章 fathonUtils 工具与窗口设计 预处理与窗口选择往往决定 DFA 结果的成败。本章详解 全部辅助函数,并给出窗口设计的可操作规则。 8.1 fathonUtils 模块概览 函数 | 作用 | 去均值,不累积 | 去均值 + 累积和 → profile | 等差窗口数组 | 等分 count 个窗口 | 等比(对数均匀)窗口 | 等比 count 个窗口 | 读取已保存对象的成员 8.
预处理与窗口选择往往决定 DFA 结果的成败。本章详解
fathonUtils全部辅助函数,并给出窗口设计的可操作规则。
from fathon import fathonUtils as fu
| 函数 | 作用 |
|---|---|
subtractMean(tsVec) |
去均值,不累积 |
toAggregated(tsVec) |
去均值 + 累积和 → profile |
linRangeByStep(start, stop, step=1) |
等差窗口数组 |
linRangeByCount(start, stop, count) |
等分 count 个窗口 |
powRangeByStep(start, stop, step, base=2) |
等比(对数均匀)窗口 |
powRangeByCount(start, stop, count, base=2) |
等比 count 个窗口 |
getObjectMember(obj, memberName) |
读取已保存对象的成员 |
import numpy as np from fathon import fathonUtils as fu x = np.array([1.0, 2.0, 3.0, 4.0, 5.0]) y = fu.subtractMean(x) # 结果:[-2, -1, 0, 1, 2] 均值为 0,但未累积
profile = fu.toAggregated(x) # 等价于 np.cumsum(x - np.mean(x))
数据类型建议:
| 原始数据 | 推荐处理 |
|---|---|
| 收益率 r(t) | 直接 toAggregated(r) |
| 价格 P(t) | 先 np.diff(np.log(P)) 得对数收益,再 toAggregated |
| 已去均值增量 | toAggregated |
| 带趋势的价格 | 先差分/去趋势,再 toAggregated |
wins = fu.linRangeByStep(10, 2000) # 10,11,...,2000 共 1991 个 wins = fu.linRangeByStep(10, 2000, step=10) # 10,20,...,2000 共 200 个
适用:希望在线性尺度上均匀采样窗口。
wins = fu.linRangeByCount(10, 2000, count=50) # 在 [10, 2000] 间均匀取 50 个点(含端点)
适用:预先固定窗口个数,便于控制计算量。
标度分析常在 log-log 空间作图,对数均匀的窗口更符合视觉与拟合需求:
# 从 2^4=16 到 2^11=2048,指数步长 0.5 wins = fu.powRangeByStep(4, 11, step=0.5, base=2) # 约:16, 22.6, 32, 45.3, ..., 2048 wins = fu.powRangeByCount(4, 11, count=30, base=2) # 30 个对数均匀窗口
| 方式 | 优点 |
|---|---|
| linRange | 小窗口分辨率高 |
| powRange | 大跨度尺度覆盖更均匀,拟合点不扎堆 |
实践:长序列(L > 10⁴)可优先 powRangeByCount;短序列用 linRangeByStep 并限制最大 n。
设序列长度 L,profile 长度 = L。
| 建议 | 理由 |
|---|---|
| n_min ≥ 10~16 | 多项式拟合需足够点数 |
| polOrd=m 时 n_min ≥ 4(m+1) | 避免欠定 |
[
n_{max} \leq L / 4
]
保证至少 4 个完整分段(N_n ≥ 4)。更保守:n_max = L / 10。
[
N_n = \lfloor L / n \rfloor \geq 4 \quad \text{(revSeg 时 } 2N_n \geq 8\text{)}
]
fitFlucVec(nStart, nEnd) 应排除:
典型:丢弃最小 10% 和最大 10% 的 wins 点。
cost_proxy = len(wins) * len(q_list) * L # 若过大,先增大 step 或 powRangeByCount 减少 wins
序列长度 L? │ ├─ L < 1000 ──► n_max = L//5, linRange step≥5, unbiased=True │ ├─ 1000 ≤ L < 10000 ──► linRangeByStep(10, L//4, step=5) │ └─ L ≥ 10000 ──► powRangeByCount(log2(16), log2(L//4), count=40) 或 linRangeByStep(10, L//4, step=10)
从已 saveObject 的对象或当前对象读取内部成员(调试/高级用法):
import fathon from fathon import fathonUtils as fu dfa = fathon.DFA(fu.toAggregated(np.random.randn(3000))) n, F = dfa.computeFlucVec(fu.linRangeByStep(10, 500)) # 读取内部存储的 n、F(名称以实际对象为准) # member = fu.getObjectMember(dfa, 'n')
具体成员名取决于 fathon 内部实现;一般用户直接使用 computeFlucVec 返回值即可。
import numpy as np import fathon from fathon import fathonUtils as fu def prepare_dfa(increments, n_min=10, n_max_ratio=4, step=5, polOrd=1): """ 通用 DFA 预处理与窗口生成。 increments: 一维增量序列 """ L = len(increments) if L < 100: raise ValueError("序列过短,建议 L >= 100") profile = fu.toAggregated(increments) n_max = L // n_max_ratio wins = fu.linRangeByStep(n_min, n_max, step=step) dfa = fathon.DFA(profile) use_unbiased = L < 1000 n, F = dfa.computeFlucVec( wins, polOrd=polOrd, revSeg=True, unbiased=use_unbiased ) # 拟合时排除首尾各 10% 窗口点 idx_lo = max(0, len(n) // 10) idx_hi = min(len(n), len(n) - len(n) // 10) H, intercept = dfa.fitFlucVec( nStart=int(n[idx_lo]), nEnd=int(n[idx_hi - 1]) ) return dict(H=H, n=n, F=F, profile=profile, wins=wins) # 演示 np.random.seed(0) inc = np.random.randn(5000) result = prepare_dfa(inc) print(f"H = {result['H']:.4f}, 窗口数 = {len(result['wins'])}")
import numpy as np import fathon from fathon import fathonUtils as fu L = 20000 profile = fu.toAggregated(np.random.randn(L)) # 对数均匀 35 个窗口 exp_min = int(np.log2(16)) exp_max = int(np.log2(L // 4)) wins = fu.powRangeByCount(exp_min, exp_max, count=35, base=2) wins = wins.astype(int) q_list = np.arange(-3, 4, 0.25) mfdfa = fathon.MFDFA(profile) n, F = mfdfa.computeFlucVec(wins, q_list, revSeg=True) h_q, _ = mfdfa.fitFlucVec() print(f"h(2)={h_q[np.argmin(np.abs(q_list-2))]:.3f}, wins={len(wins)}, q={len(q_list)}")
| 任务 | 工具 |
|---|---|
| 构造 profile | toAggregated |
| 线性窗口 | linRangeByStep / linRangeByCount |
| 对数均匀窗口 | powRangeByStep / powRangeByCount |
| 短序列 | unbiased=True + 缩小 n_max |
| 控制 MFDFA 成本 | 减少 wins 或 q 点数 |
窗口设计是 DFA 比调 polOrd 更关键的超参数。第 9 章继续讨论多项式阶数、可视化与工程陷阱。
linRange step=1 与 step=20 的 H 与耗时。powRangeByCount 与 linRangeByCount(..., count=40) 各生成 wins,画 F(n) 对比。prepare_dfa 用于你的 CSV 数据(一列数值)。「工具函数与窗口设计」示例把本章工具函数串成可运行演示,覆盖 toAggregated / subtractMean 的差异、两种窗口生成器对比、三种窗口设计法则对 H 的影响。下面给出核心的窗口设计对比脚本:
"""窗口设计对比完整示例。 验证三条窗口设计经验法则对 Hurst 估计的影响: 1. 最小窗口 ≥ 10(保证段内多项式拟合有样本) 2. 最大窗口 ≤ L/4(保证至少 4 段,统计稳定) 3. 窗口数 ≥ 8~10 个(log-log 拟合才有意义) 依赖:pip install fathon numpy """ import numpy as np import fathon from fathon import fathonUtils as fu def estimate_H(profile, wins, **kw): """对给定窗口集跑 DFA,返回 H。""" dfa = fathon.DFA(profile) dfa.computeFlucVec(wins, **kw) H, _ = dfa.fitFlucVec() return H def main(): np.random.seed(0) L = 4000 profile = fu.toAggregated(np.random.randn(L)) # 三种窗口设计:最小集 / 标准集 / 对数等距 cases = [ ("最小窗口集", fu.linRangeByStep(10, 50, step=10)), ("标准集 ", fu.linRangeByStep(10, L // 4, step=20)), ("对数等距 ", fu.linRangeByCount(16, L // 4, num=15)), ] print("=== 窗口设计对比 ===") for name, wins in cases: H = estimate_H(profile, wins, revSeg=True) print(f" {name} 窗口数={len(wins):3d} H={H:.4f}") # ── toAggregated 验证:去均值 + 累积和 ── print("\n=== toAggregated = 去均值 + 累积和 ===") x = np.random.randn(10) + 5.0 Y = fu.toAggregated(x) Y_manual = np.cumsum(x - np.mean(x)) print(f" 工具 vs 手算 max 差异: {np.max(np.abs(Y - Y_manual)):.2e}") if __name__ == "__main__": main()
💡 窗口设计是 DFA 比调 polOrd 更关键的超参数。运行窗口设计对比,看不同窗口集对同一序列 H 估计的波动,建立「窗口选错,H 不可信」的直觉。