import akshare as ak
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import time
# 字体设置
plt.rcParams["font.family"] = ["simkai.ttf", "Microsoft YaHei", "SimHei"]
try:
import matplotlib.font_manager as fm
font = fm.FontProperties(fname=r"C:\Windows\Fonts\msyh.ttc")
plt.rcParams["font.family"] = font.get_name()
except Exception:
plt.rcParams["font.family"] = ["SimHei", "simkai.ttf"]
from requests.exceptions import ConnectionError, Timeout, TooManyRedirects
def get_stock_data(stock_code, start_date, end_date, max_retries=3, retry_delay=5):
"""获取股票数据，添加重试机制和错误处理"""
for attempt in range(max_retries):
try:
print(f"尝试获取 {stock_code} 的数据，尝试次数: {attempt + 1}/{max_retries}")
stock_df = ak.stock_zh_a_hist_tx(
symbol=stock_code,
start_date=start_date,
end_date=end_date,
adjust=""
)
if stock_df is None or stock_df.empty:
print(f"警告：获取到的 {stock_code} 数据为空")
return None
if 'date' in stock_df.columns:
stock_df = stock_df.set_index('date')
stock_df.index = pd.to_datetime(stock_df.index)
else:
print(f"错误：未找到日期列。可用列名: {stock_df.columns.tolist()}")
return None
return stock_df['close']
except (ConnectionError, Timeout, TooManyRedirects) as e:
print(f"网络连接错误: {e}")
if attempt < max_retries - 1:
print(f"将在 {retry_delay} 秒后重试...")
time.sleep(retry_delay)
else:
print("达到最大重试次数，获取数据失败")
return None
except Exception as e:
print(f"发生未知错误: {e}")
return None
def monte_carlo_simulations(close_prices, num_simulations=10000, days=1):
"""返回蒙特卡洛模拟的所有收益率数组"""
returns = close_prices.pct_change().dropna()
mean_return = returns.mean()
std_return = returns.std()
simulations = np.zeros(num_simulations)
for i in range(num_simulations):
daily_returns = np.random.normal(mean_return, std_return, days)
final_return = np.prod(1 + daily_returns) - 1
simulations[i] = final_return
return simulations
def monte_carlo_var(simulations, confidence_level=0.95):
"""根据模拟结果计算VaR"""
var = np.percentile(simulations, 100 * (1 - confidence_level))
return var
def plot_var_results(simulations, var_value, confidence_level=0.95, days=1):
"""绘制蒙特卡洛模拟收益率分布和VaR"""
plt.figure(figsize=(10, 6))
sns.histplot(simulations, bins=30, kde=True, stat="density", color='skyblue')
plt.axvline(x=var_value, color='red', linestyle='--',
label=f"{int(confidence_level*100)}% VaR: {var_value:.2%}")
plt.title(f"蒙特卡洛模拟股票收益率分布与VaR（{days}天）")
plt.xlabel("收益率")
plt.ylabel("密度")
plt.legend()
plt.grid(True, linestyle='--', alpha=0.7)
plt.tight_layout()
plt.show()
def plot_price_history(stock_prices):
"""绘制历史价格走势"""
plt.figure(figsize=(10, 4))
sns.lineplot(data=stock_prices)
plt.title("历史股票价格走势")
plt.xlabel("日期")
plt.ylabel("价格")
plt.grid(True, linestyle='--', alpha=0.7)
plt.tight_layout()
plt.show()
if __name__ == "__main__":
# 参数设置
stock_code = 'sh600938' # 中国海油
start_date = '20240101'
end_date = '20250521'
num_simulations = 1000
days = 1
confidence_level = 0.95
# 获取股票数据
close_prices = get_stock_data(stock_code, start_date, end_date)
if close_prices is None:
print("无法获取股票数据，程序退出")
exit()
# 绘制历史价格
plot_price_history(close_prices)
# 蒙特卡洛模拟
simulations = monte_carlo_simulations(close_prices, num_simulations, days)
var = monte_carlo_var(simulations, confidence_level)
print(f"单只股票 {stock_code} 在 {confidence_level * 100:.0f}% 置信水平下，{days} 天的 VaR 为: {var:.2%}")
# 绘制模拟结果与VaR
plot_var_results(simulations, var, confidence_level, days)
