4. 历史现金流数据探索：描述性统计分析、季节性分解、趋势与周期识别、异常值检测与处理

历史现金流数据是预测模型的基石。在进入参数调优之前，必须对原始数据进行系统性探索，以理解其统计特征、时间结构及数据质量。本节将围绕四个核心任务展开：描述性统计、季节性分解、趋势与周期识别、异常值检测与处理。

4.1 描述性统计分析

描述性统计帮助我们快速掌握现金流数据的集中趋势、离散程度与分布形态。以下表格展示了某企业近5年经营活动现金流净额（单位：万元）的关键统计量：

4.2 季节性分解

现金流通常受季节性因素影响（如零售业的节假日回款、建筑业的年末结算）。季节性分解将时间序列拆解为三个成分：趋势-周期（Trend-Cycle）、季节性（Seasonal）和残差（Residual）。

常用的分解模型有两种：

• 加法模型： \( Y_t = T_t + S_t + R_t \)，适用于季节性波动幅度不随趋势变化的情况。
• 乘法模型： \( Y_t = T_t \times S_t \times R_t \)，适用于季节性波动幅度随趋势成比例变化的情况。

以下SVG图展示了加法分解的流程与各成分关系：

4.3 趋势与周期识别

趋势反映现金流的长期方向（上升、下降或持平），周期则指受宏观经济或行业景气度影响的非固定长度波动。识别方法包括：

• 移动平均平滑：使用12期（月度数据）或4期（季度数据）移动平均提取趋势-周期成分。
• HP滤波器：通过惩罚参数 \(\lambda\) 分离趋势与周期。月度数据推荐 \(\lambda=14400\)，季度数据 \(\lambda=1600\)。
• 差分法：一阶差分消除趋势，二阶差分消除线性趋势后的二次趋势。

4.4 异常值检测与处理

异常值会严重扭曲统计量、误导模型参数估计。现金流数据中的异常值通常源于：大额一次性收款/付款、会计调整、数据录入错误等。

4.4.1 检测方法对比

4.4.2 处理策略

检测到异常值后，需根据业务背景选择处理方式：

• 保留并标记：若异常值反映真实业务事件（如并购、资产出售），应保留并在模型中添加虚拟变量。
• 截尾（Winsorize）：将异常值替换为指定百分位（如1%和99%）的值。
• 插补：使用前后均值、中位数或线性插值替换。
• 剔除：仅当确认数据录入错误且无法修正时使用。

以下SVG图展示了异常值检测与处理的完整决策流程：

4.5 综合代码示例

以下Python代码整合了本节核心操作，使用 statsmodels 和 scipy 库完成探索性分析：

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from statsmodels.tsa.seasonal import seasonal_decompose
from scipy import stats

# 加载数据（示例：月度现金流）
df = pd.read_csv('cash_flow_monthly.csv', parse_dates=['date'], index_col='date')
cf = df['operating_cf']

# 1. 描述性统计
desc = cf.describe()
skew = cf.skew()
kurt = cf.kurtosis()
print(f"偏度: {skew:.2f}, 峰度: {kurt:.2f}")

# 2. 季节性分解（加法模型）
decomp = seasonal_decompose(cf, model='additive', period=12)
trend = decomp.trend
seasonal = decomp.seasonal
residual = decomp.resid

# 3. 异常值检测（IQR法）
Q1 = cf.quantile(0.25)
Q3 = cf.quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR
outliers = cf[(cf < lower_bound) | (cf > upper_bound)]
print(f"检测到 {len(outliers)} 个异常值")

# 4. 异常值处理（截尾到1%和99%分位）
lower_perc = cf.quantile(0.01)
upper_perc = cf.quantile(0.99)
cf_winsorized = cf.clip(lower=lower_perc, upper=upper_perc)

# 5. 平稳性检验（ADF）
from statsmodels.tsa.stattools import adfuller
adf_result = adfuller(cf.dropna())
print(f"ADF统计量: {adf_result[0]:.4f}, p值: {adf_result[1]:.4f}")