当前位置： 首页 > article >正文

python进行数据分析（以A 股为例）

article 2025/3/31 11:20:35

1. 概述

该代码使用 Dash 框架构建了一个交互式的 A 股股票分析看板应用。用户可以输入股票代码、选择日期范围，并通过按钮更新数据。应用提供了两个主要的分析视图：价格走势和技术指标，其中技术指标支持用户选择不同的指标进行展示，如 MACD、RSI、布林带和均线系统等。应用通过 AKShare 库获取 A 股股票的历史数据，并使用 Plotly 进行数据可视化。

2. 依赖库导入

python

import dash
from dash import html, dcc
from dash.dependencies import Input, Output
import pandas as pd
import numpy as np
import akshare as ak
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from datetime import datetime, timedelta
import time

  • dash：用于构建交互式 Web 应用的框架。
  • html和dcc：Dash 中的 HTML 和核心组件库，用于构建应用的用户界面。
  • Input和Output：用于定义 Dash 应用中回调函数的输入和输出。
  • pandas：用于数据处理和分析，如数据读取、清洗、转换等。
  • numpy：用于数值计算，如数组操作、数学函数等。
  • akshare：用于获取 A 股股票数据的金融数据接口库。
  • plotly.graph_objects：用于创建 Plotly 图形对象，进行数据可视化。
  • make_subplots：用于创建包含多个子图的 Plotly 图形。
  • datetime和timedelta：用于处理日期和时间相关的操作。
  • time：用于处理时间相关的操作，如时间延迟等。

3. 初始化 Dash 应用

python

app = dash.Dash(__name__)

创建一个 Dash 应用实例，__name__是 Python 的特殊变量，用于指定应用的名称。

4. 定义应用布局

python

app.layout = html.Div([
    html.H1("A股股票分析看板", style={'text-align': 'center'}),
    
    html.Div([
        dcc.Input(id='ticker-input', value='600519', type='text'),  # 茅台股票代码
        dcc.DatePickerRange(
            id='date-range',
            start_date='2023-01-01',
            end_date=datetime.now().strftime('%Y-%m-%d')
        ),
        html.Button('更新数据', id='update-button', n_clicks=0)
    ], style={'padding': '10px', 'background-color': '#f9f9f9'}),
    
    dcc.Tabs([
        dcc.Tab(label='价格走势', children=[
            dcc.Graph(id='price-chart')
        ]),
        dcc.Tab(label='技术指标', children=[
            dcc.Dropdown(
                id='indicator-dropdown',
                options=[
                    {'label': 'MACD', 'value': 'MACD'},
                    {'label': 'RSI', 'value': 'RSI'},
                    {'label': '布林带', 'value': 'BBANDS'},
                    {'label': '均线系统', 'value': 'MA'}
                ],
                value=['MACD', 'RSI'],
                multi=True
            ),
            dcc.Graph(id='indicator-chart')
        ])
    ])
])

  • html.H1：创建一个一级标题，显示应用的名称，居中对齐。
  • html.Div：包含用户输入股票代码的输入框、日期选择范围组件和更新数据按钮。
    • dcc.Input：输入框，默认显示茅台股票代码600519，用户可以输入其他股票代码。
    • dcc.DatePickerRange：日期选择范围组件，默认起始日期为2023-01-01，结束日期为当前日期。
    • html.Button：更新数据按钮，初始点击次数为 0。
  • dcc.Tabs：创建一个选项卡组件，包含两个选项卡：价格走势和技术指标。
    • dcc.Tab（价格走势）：包含一个dcc.Graph组件，用于显示股票价格走势。
    • dcc.Tab（技术指标）：包含一个dcc.Dropdown组件和一个dcc.Graph组件。dcc.Dropdown用于选择要显示的技术指标，默认选择MACD和RSI，支持多选；dcc.Graph用于显示所选技术指标的图表。

5. 获取股票数据函数

python

def get_stock_data(ticker, start_date, end_date):
    """使用AKShare获取A股股票数据"""
    try:
        # 转换日期格式
        start = pd.to_datetime(start_date).strftime('%Y%m%d')
        end = pd.to_datetime(end_date).strftime('%Y%m%d')
        
        # 获取数据
        df = ak.stock_zh_a_hist(symbol=ticker, period="daily", start_date=start, end_date=end)
        
        # 处理数据格式
        df['日期'] = pd.to_datetime(df['日期'])
        df = df.set_index('日期')
        df = df.rename(columns={
            '开盘': 'Open',
            '收盘': 'Close',
            '最高': 'High',
            '最低': 'Low',
            '成交量': 'Volume'
        })
        return df.sort_index()
    except Exception as e:
        print(f"获取数据失败: {e}")
        return pd.DataFrame()

  • ticker：股票代码。
  • start_date：起始日期。
  • end_date：结束日期。
  • 函数首先将输入的日期格式转换为 AKShare 要求的格式（%Y%m%d）。
  • 使用ak.stock_zh_a_hist函数从 AKShare 获取指定股票在指定日期范围内的每日历史数据。
  • 对获取到的数据进行格式处理，将日期列转换为datetime类型并设置为索引，同时将列名从中文转换为英文，最后按日期索引排序。
  • 如果获取数据过程中出现异常，打印错误信息并返回一个空的DataFrame。

6. 更新图表回调函数

python

@app.callback(
    [Output('price-chart', 'figure'),
     Output('indicator-chart', 'figure')],
    [Input('update-button', 'n_clicks')],
    [dash.dependencies.State('ticker-input', 'value'),
     dash.dependencies.State('date-range','start_date'),
     dash.dependencies.State('date-range', 'end_date'),
     dash.dependencies.State('indicator-dropdown', 'value')]
)
def update_charts(n_clicks, ticker, start_date, end_date, indicators):
    if not indicators:
        indicators = ['MACD', 'RSI']
    
    # 获取股票数据
    df = get_stock_data(ticker, start_date, end_date)
    
    if df.empty:
        return go.Figure(), go.Figure()
    
    # 价格走势图
    price_fig = go.Figure()
    price_fig.add_trace(go.Candlestick(x=df.index,
                                      open=df['Open'],
                                      high=df['High'],
                                      low=df['Low'],
                                      close=df['Close'],
                                      name='K线'))
    price_fig.update_layout(title=f'{ticker} 价格走势',
                          xaxis_title='日期',
                          yaxis_title='价格',
                          xaxis_rangeslider_visible=False)
    
    # 技术指标图
    indicator_fig = make_subplots(rows=2, cols=1, shared_xaxes=True, vertical_spacing=0.02)
    indicator_fig.add_trace(go.Candlestick(x=df.index,
                                         open=df['Open'],
                                         high=df['High'],
                                         low=df['Low'],
                                         close=df['Close'],
                                         name='K线'), row=1, col=1)
    
    # 计算并添加技术指标
    if 'MACD' in indicators:
        exp12 = df['Close'].ewm(span=12, adjust=False).mean()
        exp26 = df['Close'].ewm(span=26, adjust=False).mean()
        macd = exp12 - exp26
        signal = macd.ewm(span=9, adjust=False).mean()
        
        indicator_fig.add_trace(go.Scatter(x=df.index, y=macd, name='MACD'), row=2, col=1)
        indicator_fig.add_trace(go.Scatter(x=df.index, y=signal, name='信号线'), row=2, col=1)
    
    if 'RSI' in indicators:
        delta = df['Close'].diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
        rs = gain / loss
        rsi = 100 - (100 / (1 + rs))
        
        indicator_fig.add_trace(go.Scatter(x=df.index, y=rsi, name='RSI'), row=2, col=1)
        indicator_fig.add_hline(y=70, line_dash="dash", line_color="red", row=2, col=1)
        indicator_fig.add_hline(y=30, line_dash="dash", line_color="green", row=2, col=1)
    
    indicator_fig.update_layout(title=f'{ticker} 技术指标分析',
                              height=800,
                              xaxis_rangeslider_visible=False)
    
    return price_fig, indicator_fig

  • 该回调函数在用户点击更新数据按钮时触发。
  • 输入参数：
    • n_clicks：更新数据按钮的点击次数。
    • ticker：用户输入的股票代码。
    • start_date：用户选择的起始日期。
    • end_date：用户选择的结束日期。
    • indicators：用户选择的技术指标列表。
  • 输出参数：
    • price-chart的figure：价格走势图。
    • indicator-chart的figure：技术指标图。
  • 函数首先检查indicators是否为空，如果为空则设置默认值为['MACD', 'RSI']。
  • 调用get_stock_data函数获取股票数据，如果数据为空则返回两个空的Figure对象。
  • 价格走势图：创建一个go.Figure对象，添加一个 K 线图迹，设置图表标题、X 轴和 Y 轴标题，并隐藏 X 轴的范围滑块。
  • 技术指标图：创建一个包含两个子图的make_subplots对象，第一个子图添加 K 线图迹，第二个子图根据用户选择的技术指标计算并添加相应的迹。
    • MACD指标：计算 12 日和 26 日的指数移动平均线，然后计算 MACD 值和信号线值，并添加到第二个子图中。
    • RSI指标：计算价格的差值、平均收益和平均损失，进而计算 RSI 值，并添加到第二个子图中，同时添加 70 和 30 的水平线作为超买和超卖的参考。
  • 最后设置技术指标图的标题、高度，并隐藏 X 轴的范围滑块，返回价格走势图和技术指标图。

7. 运行应用

python

if __name__ == '__main__':
    app.run(debug=True)

当脚本直接运行时，启动 Dash 应用，并设置debug模式为True，以便在开发过程中查看错误信息和自动重新加载应用。

完整代码

import dash
from dash import html, dcc
from dash.dependencies import Input, Output
import pandas as pd
import numpy as np
import akshare as ak
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from datetime import datetime, timedelta
import time

# 初始化Dash应用
app = dash.Dash(__name__)

app.layout = html.Div([
    html.H1("A股股票分析看板", style={'text-align': 'center'}),
    
    html.Div([
        dcc.Input(id='ticker-input', value='600519', type='text'),  # 茅台股票代码
        dcc.DatePickerRange(
            id='date-range',
            start_date='2023-01-01',
            end_date=datetime.now().strftime('%Y-%m-%d')
        ),
        html.Button('更新数据', id='update-button', n_clicks=0)
    ], style={'padding': '10px', 'background-color': '#f9f9f9'}),
    
    dcc.Tabs([
        dcc.Tab(label='价格走势', children=[
            dcc.Graph(id='price-chart')
        ]),
        dcc.Tab(label='技术指标', children=[
            dcc.Dropdown(
                id='indicator-dropdown',
                options=[
                    {'label': 'MACD', 'value': 'MACD'},
                    {'label': 'RSI', 'value': 'RSI'},
                    {'label': '布林带', 'value': 'BBANDS'},
                    {'label': '均线系统', 'value': 'MA'}
                ],
                value=['MACD', 'RSI'],
                multi=True
            ),
            dcc.Graph(id='indicator-chart')
        ])
    ])
])

def get_stock_data(ticker, start_date, end_date):
    """使用AKShare获取A股股票数据"""
    try:
        # 转换日期格式
        start = pd.to_datetime(start_date).strftime('%Y%m%d')
        end = pd.to_datetime(end_date).strftime('%Y%m%d')
        
        # 获取数据
        df = ak.stock_zh_a_hist(symbol=ticker, period="daily", start_date=start, end_date=end)
        
        # 处理数据格式
        df['日期'] = pd.to_datetime(df['日期'])
        df = df.set_index('日期')
        df = df.rename(columns={
            '开盘': 'Open',
            '收盘': 'Close',
            '最高': 'High',
            '最低': 'Low',
            '成交量': 'Volume'
        })
        return df.sort_index()
    except Exception as e:
        print(f"获取数据失败: {e}")
        return pd.DataFrame()

@app.callback(
    [Output('price-chart', 'figure'),
     Output('indicator-chart', 'figure')],
    [Input('update-button', 'n_clicks')],
    [dash.dependencies.State('ticker-input', 'value'),
     dash.dependencies.State('date-range', 'start_date'),
     dash.dependencies.State('date-range', 'end_date'),
     dash.dependencies.State('indicator-dropdown', 'value')]
)
def update_charts(n_clicks, ticker, start_date, end_date, indicators):
    if not indicators:
        indicators = ['MACD', 'RSI']
    
    # 获取股票数据
    df = get_stock_data(ticker, start_date, end_date)
    
    if df.empty:
        return go.Figure(), go.Figure()
    
    # 价格走势图
    price_fig = go.Figure()
    price_fig.add_trace(go.Candlestick(x=df.index,
                                      open=df['Open'],
                                      high=df['High'],
                                      low=df['Low'],
                                      close=df['Close'],
                                      name='K线'))
    price_fig.update_layout(title=f'{ticker} 价格走势',
                          xaxis_title='日期',
                          yaxis_title='价格',
                          xaxis_rangeslider_visible=False)
    
    # 技术指标图
    indicator_fig = make_subplots(rows=2, cols=1, shared_xaxes=True, vertical_spacing=0.02)
    indicator_fig.add_trace(go.Candlestick(x=df.index,
                                         open=df['Open'],
                                         high=df['High'],
                                         low=df['Low'],
                                         close=df['Close'],
                                         name='K线'), row=1, col=1)
    
    # 计算并添加技术指标
    if 'MACD' in indicators:
        exp12 = df['Close'].ewm(span=12, adjust=False).mean()
        exp26 = df['Close'].ewm(span=26, adjust=False).mean()
        macd = exp12 - exp26
        signal = macd.ewm(span=9, adjust=False).mean()
        
        indicator_fig.add_trace(go.Scatter(x=df.index, y=macd, name='MACD'), row=2, col=1)
        indicator_fig.add_trace(go.Scatter(x=df.index, y=signal, name='信号线'), row=2, col=1)
    
    if 'RSI' in indicators:
        delta = df['Close'].diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
        rs = gain / loss
        rsi = 100 - (100 / (1 + rs))
        
        indicator_fig.add_trace(go.Scatter(x=df.index, y=rsi, name='RSI'), row=2, col=1)
        indicator_fig.add_hline(y=70, line_dash="dash", line_color="red", row=2, col=1)
        indicator_fig.add_hline(y=30, line_dash="dash", line_color="green", row=2, col=1)
    
    indicator_fig.update_layout(title=f'{ticker} 技术指标分析',
                              height=800,
                              xaxis_rangeslider_visible=False)
    
    return price_fig, indicator_fig

if __name__ == '__main__':
    app.run(debug=True)


http://www.kler.cn/a/611915.html

相关文章：

  • uniapp开发中store的基本用法和模块化详解
  • uniapp上传图片之前压缩处理
  • 解构 HarmonyOS：技术神话背后的理性审视
  • 给聊天机器人装“短期记忆“：Flask版实现指南
  • 高级数据结构02AVL树
  • AI编程工具哪家强？对比Cusor、Copilot、Cline
  • C++算法深度解析：快速排序的实现、分析与优化
  • RoMA: 基于Mamba的遥感基础模型, 已开源, 首次验证mamba的scaling能力
  • 用数组遍历出来的页面，随节点创建的ref存储在数据仓库中，如果数据删除，页面相关节点也会删除，数据仓库中随节点创建的ref会不会也同时删除
  • STM32F103C8T6移植DMP解算MPU6050
  • Elasticsearch：使用 Azure AI 文档智能解析 PDF 文本和表格数据
  • linux---------进程概念(完)
  • 字节真题，问a,b,c指的地址是否相同？
  • SQL Server常见问题解析
  • 记录react和vue 属性传组件相关的事宜
  • 微软重磅发布 OmniParser V2.0：AI 视觉解析能力跃升，开启界面自动化新时代
  • 鸿蒙Flutter实战：20. Flutter集成高德地图，同层渲染
  • AG7220替代方案|ASL6328芯片设计|HDMI2.0 Retimer中继器方案
  • 核函数（机器学习深度学习）
  • win11+ubuntu双系统安装