1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
| import numpy as np
import pandas as pd
from statsmodels.tsa.stattools import adfuller, kpss
# 1. 计算对数收益率
df['对数收益率'] = np.log(df['成交价'] / df['成交价'].shift(1))
# 删除第一个NaN值（因为shift操作产生的）
df = df.dropna(subset=['对数收益率'])
# 2. 对处理后的数据进行ADF检验
def adf_test(series):
"""执行ADF检验并格式化结果"""
result = adfuller(series.dropna())
adf_stat = result[0]
p_value = result[1]
critical_values = result[4]
# 判断结论
conclusion = "拒绝原假设，序列平稳" if p_value < 0.05 else "无法拒绝原假设，序列非平稳"
return pd.Series({
'ADF统计量': round(adf_stat, 4),
'p值': round(p_value, 4),
'1%临界值': round(critical_values['1%'], 4),
'5%临界值': round(critical_values['5%'], 4),
'10%临界值': round(critical_values['10%'], 4),
'结论': conclusion
})
# 执行ADF检验
adf_result_log = adf_test(df['对数收益率'])
print("对数收益率的ADF检验结果：")
print(adf_result_log.to_string())
# 3. 对处理后的数据进行KPSS检验
def kpss_test(series, regression_type='c'):
"""执行KPSS检验并格式化结果"""
result = kpss(series, regression=regression_type, nlags='auto')
statistic = result[0]
p_value = result[1]
critical_values = result[3]
# 判断结论
conclusion = "拒绝原假设，序列非平稳" if p_value < 0.05 else "无法拒绝原假设，序列平稳"
return pd.Series({
'KPSS统计量': round(statistic, 4),
'p值': round(p_value, 4),
'10%临界值': round(critical_values['10%'], 4),
'5%临界值': round(critical_values['5%'], 4),
'2.5%临界值': round(critical_values['2.5%'], 4),
'1%临界值': round(critical_values['1%'], 4),
'结论': conclusion
})
# 执行KPSS检验（金融数据推荐使用趋势平稳性检验）
kpss_result_log = kpss_test(df['对数收益率'], regression_type='ct')
print("\n对数收益率的KPSS检验结果：")
print(kpss_result_log.to_string())
# 4. 可视化原始价格与对数收益率
import matplotlib.pyplot as plt
plt.figure(figsize=(14, 10))
# 原始价格序列
plt.subplot(2, 1, 1)
plt.plot(df.index, df['成交价'], 'b-')
plt.title('原始成交价序列')
plt.ylabel('价格')
plt.grid(alpha=0.3)
# 对数收益率序列
plt.subplot(2, 1, 2)
plt.plot(df.index, df['对数收益率'], 'g-')
plt.title('对数收益率序列')
plt.ylabel('对数收益率')
plt.grid(alpha=0.3)
plt.tight_layout()
plt.show()
# 5. 综合判断平稳性（结合ADF和KPSS结果）
def combined_stationarity(adf_result, kpss_result):
"""综合判断平稳性结果"""
adf_stationary = adf_result['结论'].startswith("拒绝")
kpss_stationary = kpss_result['结论'].startswith("无法")
if adf_stationary and kpss_stationary:
return "强平稳证据：ADF和KPSS结果一致表明序列平稳"
elif not adf_stationary and not kpss_stationary:
return "强非平稳证据：ADF和KPSS结果一致表明序列非平稳"
elif not adf_stationary and kpss_stationary:
return "趋势平稳：建议进行去趋势处理"
else:
return "差分平稳：建议进行差分处理"
# 输出综合结论
conclusion = combined_stationarity(adf_result_log, kpss_result_log)
print("\n平稳性综合结论：", conclusion)
|
