绘制某只股票的K线图、收益率、计算VaR并绘图

# -*- coding: utf-8 -*-
"""
获取某只股票的数据绘制K线图、计算日收益率、绘制收益率直方图、计算VaR
并绘制VaR图 
Created on Sun Sep 15 13:15:50 2024
@author: 24384
"""

import matplotlib.pyplot as plt
import matplotlib.font_manager as fm
import akshare as ak
import numpy as np
import pandas as pd
import mplfinance as mpf  # 需先 pip install mplfinance
import seaborn as sns
import time
from requests.exceptions import ConnectionError, Timeout, TooManyRedirects

# # 列出系统所有可用字体
# fonts = fm.findSystemFonts()
# font_names = [fm.FontProperties(fname=font).get_name() for font in fonts]
# print("系统可用字体:")
# for font_name in font_names:
#     print(font_name)

# 手动选择一个支持中文的字体
font_name = 'Microsoft YaHei'  # 根据输出结果修改
try:
    plt.rcParams['font.family'] = font_name
    plt.rcParams['axes.unicode_minus'] = False  # 负号正常显示
    print(f"成功使用字体: {font_name}")
except Exception as e:
    print(f"加载字体时发生错误: {e}，使用默认字体。")

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 plot_kline(stock_code, start_date, end_date):
    try:
        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("K线数据为空，无法绘制K线图")
            return
        stock_df = stock_df.rename(columns={'date': 'Date', 'open': 'Open', 'high': 'High', 'low': 'Low', 'close': 'Close', 'amount': 'Volume'})
        stock_df['Date'] = pd.to_datetime(stock_df['Date'])
        stock_df.set_index('Date', inplace=True)
        title = f'{stock_code} K线图'
        mpf.plot(stock_df[['Open', 'High', 'Low', 'Close', 'Volume']], type='candle', volume=True, style='yahoo', title=title)
    except RequestException as e:
        print(f"获取数据时发生网络请求错误: {e}")
    except Exception as e:
        print(f"绘制K线图时发生错误: {e}")


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 = 10000
    days = 10
    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)
    # 绘制K线图
    plot_kline(stock_code, start_date, end_date)