数据可视化

        可交互界面设计的不错，颜色搭配效果很好，不仅添加了时间区间设置功能，还可以任意选择股票代码查看股票行情的历史数据，鼠标移动至折线图上会有详细的数据展示。

 

下面给出股票数据可视化的部分代码：
# -*- coding: utf-8 -*-
import os
import pandas as pd
import akshare as ak
from datetime import datetime
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed
import matplotlib.pyplot as plt
import matplotlib.font_manager as fm
 
# ==================== 用户配置部分 ====================
# A股股票代码列表（示例：贵州茅台、宁德时代）
a_stock_symbols = ["600519", "300750","601899"]  # 支持数字或带后缀的代码（如600519.SH）
 
# 时间范围
start_date = "2020-01-01"
end_date = datetime.now().strftime("%Y-%m-%d")
 
# 数据保存路径
save_dir = "./stock_data"  # 数据存储主目录
a_stock_dir = os.path.join(save_dir, "A股数据")  # A股数据子目录
plot_dir = os.path.join(save_dir, "可视化图表")  # 可视化图表目录
 
# 创建目录（如果不存在）
os.makedirs(a_stock_dir, exist_ok=True)
os.makedirs(plot_dir, exist_ok=True)
 
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[
        logging.FileHandler(os.path.join(save_dir, "stock_data.log")),  # 日志文件
        logging.StreamHandler()  # 控制台输出
    ]
)
 
# 设置中文字体
try:
    # Windows 系统使用黑体
    plt.rcParams["font.sans-serif"] = ["SimHei"]
    # macOS 或 Linux 系统使用 Arial Unicode MS 或文泉驿字体
    # plt.rcParams["font.sans-serif"] = ["Arial Unicode MS"]  # macOS
    # plt.rcParams["font.sans-serif"] = ["WenQuanYi Zen Hei"]  # Linux
    plt.rcParams["axes.unicode_minus"] = False  # 解决负号显示问题
except Exception as e:
    logging.error(f"字体设置失败: {e}")
 
# ======================================================
 
def fetch_a_stock_data(symbol, retries=3):
    """通过AKShare获取单只A股数据（后复权）"""
    for attempt in range(retries):
        try:
            # 去掉股票代码的后缀（如 .SH 或 .SZ）
            symbol_clean = symbol.split(".")[0]
 
            # 获取数据
            df = ak.stock_zh_a_hist(symbol=symbol_clean, period="daily", adjust="hfq")
            if df.empty:
                logging.warning(f"[A股] 数据为空（{symbol_clean}）")
                return None
 
            # 重命名列
            df.rename(columns={
                "日期": "Date",
                "开盘": "Open",
                "收盘": "Close",
                "最高": "High",
                "最低": "Low",
                "成交量": "Volume"
            }, inplace=True)
 
            # 转换日期格式并设置为索引
            df["Date"] = pd.to_datetime(df["Date"])
            df.set_index("Date", inplace=True)
 
            logging.info(f"[A股] 数据获取成功（{symbol_clean}）")
            return df
        except Exception as e:
            if attempt < retries - 1:
                logging.warning(f"[A股] 数据获取失败（{symbol_clean}），第 {attempt + 1} 次重试...")
            else:
                logging.error(f"[A股] 数据获取失败（{symbol_clean}）: {e}")
                return None
 
def clean_data(df):
    """数据清洗：处理缺失值和异常值"""
    if df is None or df.empty:
        return None
 
    # 1. 删除完全缺失的行
    df_clean = df.dropna(how="all")
 
    # 2. 处理零成交量（视为停牌/无效数据）
    df_clean = df_clean[df_clean["Volume"] > 0]
 
    # 3. 处理价格异常值（如价格<=0或单日涨跌幅超过50%）
    price_columns = ["Open", "High", "Low", "Close"]
    for col in price_columns:
        df_clean = df_clean[df_clean[col] > 0]  # 删除价格为负或零的异常记录
 
    # 检查清洗后的数据是否为空
    if df_clean.empty:
        logging.warning("数据清洗后为空，可能全部为异常值")
        return None
 
    return df_clean
 
def save_to_csv(df, symbol):
    """保存清洗后的数据到CSV文件"""
    if df is None or df.empty:
        logging.warning(f"无有效数据可保存（{symbol}）")
        return
 
    # 保存路径
    save_path = os.path.join(a_stock_dir, f"{symbol}.csv")
    df.to_csv(save_path, encoding="utf-8-sig")  # 兼容中文路径
    logging.info(f"数据已保存至：{save_path}")
 

2数据可视化

3分析

        该代码的主要功能是通过akshare库获取A股股票的历史数据，进行数据清洗、保存和可视化。

函数解释
函数    解释
fetch_a_stock_data    获取单只股票的历史数据
clean_data    清洗数据，处理缺失值和异常值
save_to_csv    将清洗后的数据保存为CSV文件
plot_stock_data    绘制股票价格走势图并保存为PNG文件
process_single_stock    处理单只股票的完整流程（获取、清洗、保存、可视化）
batch_process_a_stocks    批量处理所有股票代码（使用多线程）
————————————————

功能丰富的静态行情图

1代码

        除了收盘价以外，行情图添加了5日均线、10日均线和20日均线，包括不同时间段的成交量数据，下面给出股票行情数据可视化的部分代码：

# ======================================================
def fetch_a_stock_data(symbol, retries=3):
    """通过AKShare获取单只A股数据（后复权）"""
    for attempt in range(retries):
        try:
            # 去掉股票代码的后缀（如 .SH 或 .SZ）
            symbol_clean = symbol.split(".")[0]
 
            # 获取数据
            df = ak.stock_zh_a_hist(symbol=symbol_clean, period="daily", adjust="hfq")
            if df.empty:
                logging.warning(f"[A股] 数据为空（{symbol_clean}）")
                return None
 
            # 重命名列
            df.rename(columns={
                "日期": "Date",
                "开盘": "Open",
                "收盘": "Close",
                "最高": "High",
                "最低": "Low",
                "成交量": "Volume"
            }, inplace=True)
 
            # 转换日期格式并设置为索引
            df["Date"] = pd.to_datetime(df["Date"])
            df.set_index("Date", inplace=True)
 
            logging.info(f"[A股] 数据获取成功（{symbol_clean}）")
            return df
        except Exception as e:
            if attempt < retries - 1:
                logging.warning(f"[A股] 数据获取失败（{symbol_clean}），第 {attempt + 1} 次重试...")
            else:
                logging.error(f"[A股] 数据获取失败（{symbol_clean}）: {e}")
                return None
 
def clean_data(df):
    """数据清洗：处理缺失值和异常值"""
    if df is None or df.empty:
        return None
 
    # 1. 删除完全缺失的行
    df_clean = df.dropna(how="all")
 
    # 2. 处理零成交量（视为停牌/无效数据）
    df_clean = df_clean[df_clean["Volume"] > 0]
 
    # 3. 处理价格异常值（如价格<=0或单日涨跌幅超过50%）
    price_columns = ["Open", "High", "Low", "Close"]
    for col in price_columns:
        df_clean = df_clean[df_clean[col] > 0]  # 删除价格为负或零的异常记录
 
    # 检查清洗后的数据是否为空
    if df_clean.empty:
        logging.warning("数据清洗后为空，可能全部为异常值")
        return None
 
    return df_clean
 
def save_to_csv(df, symbol):
    """保存清洗后的数据到CSV文件"""
    if df is None or df.empty:
        logging.warning(f"无有效数据可保存（{symbol}）")
        return
 
    # 保存路径
    save_path = os.path.join(a_stock_dir, f"{symbol}.csv")
    df.to_csv(save_path, encoding="utf-8-sig")  # 兼容中文路径
    logging.info(f"数据已保存至：{save_path}")
 
def plot_stock_data(df, symbol):
    """专业级股票行情走势图"""
    if df is None or df.empty:
        logging.warning(f"无有效数据可绘制（{symbol}）")
        return
 

2数据可视化

 

3分析

        调用batch_process_a_stocks()函数，批量处理用户配置的A股股票数据。处理完成后，记录日志信息。
 

具有可交互界面的行情图

1代码

# ==================== 布局设计 ====================
app.layout = html.Div([
    html.Div([
        html.H1("专业级股票分析系统", style={'color': 'white'}),
        html.Div([
            dcc.Input(
                id='stock-input',
                type='text',
                value='600519',
                placeholder='输入股票代码',
                style={'width': '150px', 'margin-right': '10px'}
            ),
            dcc.DatePickerRange(
                id='date-picker',
                min_date_allowed=datetime(2010, 1, 1),
                max_date_allowed=datetime.today(),
                start_date=datetime.today() - timedelta(days=365),
                end_date=datetime.today(),
                display_format='YYYY-MM-DD'
            ),
            dcc.Dropdown(
                id='indicator-selector',
                options=[
                    {'label': 'MACD', 'value': 'MACD'},
                    {'label': 'RSI', 'value': 'RSI'},
                    {'label': '布林线', 'value': 'BOLL'},
                ],
                value=['MACD'],
                multi=True,
                style={'width': '300px', 'margin-left': '10px'}
            )
        ], style={'padding': '20px', 'backgroundColor': '#1a1a1a'})
    ]),
 
    dcc.Loading(
        id="loading",
        type="circle",
        children=[
            dcc.Graph(id='main-chart', style={'height': '600px'}),
            dcc.Interval(id='interval-component', interval=60 * 1000, n_intervals=0)
        ]
    )
], style={'backgroundColor': '#1a1a1a', 'height': '100vh'})