Python 量化股票 K 线图

Python 量化股票 K 线图完整指南

K线图是技术分析的核心工具，在量化交易中用于策略可视化、形态识别和交易信号验证。Python 提供了专业K线图库，支持OHLCV数据、自定义指标、多周期分析和自动化生成。

1. 核心 K 线图库对比

库名	特点	优势	缺点
mplfinance	TradingView风格，专业金融图表	语法简洁、样式美观、指标集成	交互性较弱
Plotly	交互式K线，支持缩放拖拽	Web友好、实时更新、多图联动	性能稍差
Bokeh	高性能交互，大数据友好	实时流数据、服务器端渲染	学习曲线陡
Cufflinks	Pandas一键绘图	快速原型、集成度高	维护较少
TV-like (lightweight-charts)	TradingView原生JS引擎	专业外观、高性能	需要前端集成

2. mplfinance – 专业 K 线图绘制

基础 K 线图

import yfinance as yf
import mplfinance as mpf
import pandas as pd
from datetime import datetime

# 获取数据
ticker = "AAPL"
df = yf.download(ticker, start="2024-01-01", end="2025-01-01")

# 确保数据格式正确
df.index = pd.to_datetime(df.index)
df = df[['Open', 'High', 'Low', 'Close', 'Volume']].dropna()

# 基础K线图
mpf.plot(df, 
         type='candle',
         style='charles',  # 预设样式
         volume=True,
         title=f'{ticker} K线图',
         ylabel='价格 (USD)',
         figsize=(12, 8),
         savefig='basic_candle.png')

# 自定义样式
mc = mpf.make_marketcolors(up='g', down='r', 
                          edge='black', wick={'up':'green', 'down':'red'})
s = mpf.make_mpf_style(marketcolors=mc, gridstyle='-', y_on_right=True)
mpf.plot(df, type='candle', style=s, volume=True, figsize=(12, 8))

技术指标叠加

import talib

# 计算技术指标
df['MA20'] = df['Close'].rolling(20).mean()
df['MA60'] = df['Close'].rolling(60).mean()
df['RSI'] = talib.RSI(df['Close'], timeperiod=14)
macd, macdsignal, macdhist = talib.MACD(df['Close'])
df['MACD'] = macd
df['MACD_signal'] = macdsignal
df['MACD_hist'] = macdhist

# 绘制主图指标
apds = [
    mpf.make_addplot(df['MA20'], color='blue', width=1, panel=0),
    mpf.make_addplot(df['MA60'], color='red', width=1, panel=0),
]

# 副图指标
apdv = [
    mpf.make_addplot(df['RSI'], panel=1, color='purple', ylabel='RSI'),
    mpf.make_addplot(df['MACD'], panel=2, color='blue', ylabel='MACD'),
    mpf.make_addplot(df['MACD_signal'], panel=2, color='red'),
    mpf.make_addplot(df['MACD_hist'], type='bar', panel=2, color='gray', alpha=0.3),
]

# 水平线
hlines = dict(hlines=[70, 30], colors=['r','g'], linestyle='--', panel=1)

mpf.plot(df, type='candle', style='yahoo', volume=True,
         addplot=apds + apdv, hlines=hlines,
         title=f'{ticker} 技术分析', 
         panel_ratios=(3,1,1), figsize=(15, 10),
         savefig=f'{ticker}_analysis.png')

多股票对比

# 多只股票K线对比
tickers = ['AAPL', 'MSFT', 'GOOGL']
data = yf.download(tickers, start='2024-01-01')['Close']

mpf.plot(data, type='line', style='yahoo', volume=False,
         title='科技股价格对比', ylabel='价格 (USD)',
         figsize=(12, 6), savefig='tech_stocks.png')

# 主图显示AAPL，其他股票作为addplot
df_aapl = yf.download('AAPL', start='2024-01-01')
df_others = yf.download(['MSFT', 'GOOGL'], start='2024-01-01')['Close']

apds = [mpf.make_addplot(df_others[col], color='gray', alpha=0.5, panel=0) 
        for col in df_others.columns]

mpf.plot(df_aapl, type='candle', addplot=apds, volume=True,
         title='AAPL vs MSFT/GOOGL', figsize=(12, 8))

3. Plotly 交互式 K 线图

基础交互式K线

import plotly.graph_objects as go
from plotly.subplots import make_subplots
import pandas as pd

# 数据准备
df = yf.download('AAPL', start='2024-01-01')

fig = make_subplots(
    rows=2, cols=1, shared_xaxes=True,
    vertical_spacing=0.03,
    subplot_titles=('AAPL K线图', '成交量'),
    row_width=[0.2, 0.7],
    specs=[[{"secondary_y": False}], [{"secondary_y": False}]]
)

# K线
fig.add_trace(
    go.Candlestick(
        x=df.index,
        open=df['Open'], high=df['High'],
        low=df['Low'], close=df['Close'],
        name='AAPL',
        increasing_line_color='green', decreasing_line_color='red'
    ),
    row=1, col=1
)

# 成交量
fig.add_trace(
    go.Bar(x=df.index, y=df['Volume'], name='成交量',
           marker_color='rgba(158,202,225,0.7)'),
    row=2, col=1
)

fig.update_layout(
    title='AAPL 交互式K线图',
    yaxis_title='价格 (USD)',
    height=600,
    xaxis_rangeslider_visible=False,
    template='plotly_white'
)

fig.show()
fig.write_html('aapl_interactive.html')

高级交互式分析

# 添加技术指标的交互式图表
ma20 = df['Close'].rolling(20).mean()
ma60 = df['Close'].rolling(60).mean()
rsi = talib.RSI(df['Close'], 14)

fig = make_subplots(
    rows=4, cols=1, shared_xaxes=True,
    vertical_spacing=0.05,
    row_heights=[0.5, 0.2, 0.15, 0.15],
    subplot_titles=('K线 + 均线', 'MACD', 'RSI', '成交量')
)

# 主图：K线 + 均线
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)
fig.add_trace(go.Scatter(x=df.index, y=ma20, name='MA20', line=dict(color='blue')), row=1, col=1)
fig.add_trace(go.Scatter(x=df.index, y=ma60, name='MA60', line=dict(color='red')), row=1, col=1)

# MACD
macd_line = talib.MACD(df['Close'])[0]
macd_signal = talib.MACD(df['Close'])[1]
fig.add_trace(go.Scatter(x=df.index, y=macd_line, name='MACD', line=dict(color='blue')), row=2, col=1)
fig.add_trace(go.Scatter(x=df.index, y=macd_signal, name='Signal', line=dict(color='red')), row=2, col=1)

# RSI
fig.add_trace(go.Scatter(x=df.index, y=rsi, name='RSI', line=dict(color='purple')), row=3, col=1)
fig.add_hline(y=70, line_dash="dash", line_color="red", row=3, col=1)
fig.add_hline(y=30, line_dash="dash", line_color="green", row=3, col=1)

# 成交量
colors = ['green' if df['Close'][i] >= df['Open'][i] else 'red' for i in range(len(df))]
fig.add_trace(go.Bar(x=df.index, y=df['Volume'], name='Volume', 
                     marker_color=colors, opacity=0.6), row=4, col=1)

fig.update_layout(height=1000, title_text="AAPL 综合技术分析", showlegend=True)
fig.show()

4. 交易信号与形态标注

买卖信号可视化

import numpy as np

# 生成交易信号
df['Signal'] = 0
df['MA_cross'] = np.where(df['MA20'] > df['MA60'], 1, -1)
df['Signal'] = np.where(df['MA_cross'].diff() > 0, 1, np.where(df['MA_cross'].diff() < 0, -1, 0))

# 买卖点
buy_signals = df[df['Signal'] == 1]
sell_signals = df[df['Signal'] == -1]

# mplfinance 标注
apd_signals = [
    mpf.make_addplot(df['MA20'], color='blue'),
    mpf.make_addplot(df['MA60'], color='red'),
    # 买卖信号
    mpf.make_addplot(pd.Series(index=df.index, data=np.where(df['Signal']==1, df['Low']*0.99, np.nan)), 
                     type='scatter', markersize=100, marker='^', color='green'),
    mpf.make_addplot(pd.Series(index=df.index, data=np.where(df['Signal']==-1, df['High']*1.01, np.nan)), 
                     type='scatter', markersize=100, marker='v', color='red'),
]

mpf.plot(df, type='candle', addplot=apd_signals, volume=True,
         title='交易信号K线图', style='blueskies', figsize=(15, 10))

K线形态识别

def identify_patterns(df):
    """识别K线形态"""
    open_, high, low, close = df['Open'], df['High'], df['Low'], df['Close']

    # 使用TA-Lib识别形态
    import talib

    patterns = pd.DataFrame(index=df.index)

    # 基础形态
    patterns['Doji'] = talib.CDLDOJI(open_, high, low, close)
    patterns['Hammer'] = talib.CDLHAMMER(open_, high, low, close)
    patterns['ShootingStar'] = talib.CDLSHOOTINGSTAR(open_, high, low, close)
    patterns['Engulfing'] = talib.CDLENGULFING(open_, high, low, close)

    # 趋势形态
    patterns['MorningStar'] = talib.CDLMORNINGSTAR(open_, high, low, close)
    patterns['EveningStar'] = talib.CDLEVENINGSTAR(open_, high, low, close)

    return patterns

# 形态识别与标注
patterns = identify_patterns(df)

# 只显示有形态的日期
pattern_signals = patterns[(patterns != 0).any(axis=1)]

# Plotly标注形态
fig = go.Figure(go.Candlestick(x=df.index, open=df['Open'], high=df['High'],
                              low=df['Low'], close=df['Close']))

# 标注形态
for date, row in pattern_signals.iterrows():
    pattern_name = row[row != 0].index[0]
    fig.add_annotation(x=date, y=df.loc[date, 'High']*1.02,
                      text=pattern_name, showarrow=True,
                      arrowhead=2, arrowcolor='orange')

fig.update_layout(title='K线形态识别', height=600)
fig.show()

5. 多周期分析

多时间框架K线图

def plot_multi_timeframe(ticker, periods=['1d', '1h', '5m']):
    """多周期对比"""
    fig = make_subplots(rows=len(periods), cols=1, 
                       subplot_titles=[f'{p}周期' for p in periods],
                       vertical_spacing=0.05)

    for i, period in enumerate(periods):
        df = yf.download(ticker, period=period, interval='1h' if period=='1d' else period)

        fig.add_trace(
            go.Candlestick(x=df.index, open=df['Open'], high=df['High'],
                          low=df['Low'], close=df['Close'],
                          name=f'{period}'),
            row=i+1, col=1
        )

    fig.update_layout(height=400*len(periods), showlegend=False)
    fig.show()

plot_multi_timeframe('AAPL', ['1d', '5d'])

自适应时间框架

def resample_ohlcv(df, timeframe='1H'):
    """重采样OHLCV数据"""
    ohlc_dict = {
        'Open': 'first',
        'High': 'max',
        'Low': 'min',
        'Close': 'last',
        'Volume': 'sum'
    }
    return df.resample(timeframe).agg(ohlcv_dict).dropna()

# 生成多周期数据
df_1min = yf.download('AAPL', period='5d', interval='1m')
df_5min = resample_ohlcv(df_1min, '5T')
df_1h = resample_ohlcv(df_1min, '1H')

# 对比绘制
fig, axes = plt.subplots(3, 1, figsize=(15, 12))
for i, (data, title) in enumerate([(df_1min, '1分钟'), (df_5min, '5分钟'), (df_1h, '1小时')]):
    mpf.plot(data, type='candle', ax=axes[i], volume=False,
             style='charles', title=title, ylabel='价格')
plt.tight_layout()

6. 批量 K 线图生成

自动化报告生成

import os
from pathlib import Path
import warnings
warnings.filterwarnings('ignore')

def generate_stock_report(tickers, start_date, output_dir='kline_reports'):
    """批量生成股票K线分析报告"""
    Path(output_dir).mkdir(exist_ok=True)

    for ticker in tickers:
        try:
            print(f"处理 {ticker}...")
            df = yf.download(ticker, start=start_date)

            if df.empty:
                continue

            # 计算指标
            df['MA20'] = df['Close'].rolling(20).mean()
            df['MA60'] = df['Close'].rolling(60).mean()

            # 交易信号
            df['Signal'] = np.where(df['MA20'] > df['MA60'], 1, -1)
            buy_signals = df[df['Signal'].diff() == 2]
            sell_signals = df[df['Signal'].diff() == -2]

            # 绘制
            apds = [
                mpf.make_addplot(df['MA20'], color='blue'),
                mpf.make_addplot(df['MA60'], color='red'),
                mpf.make_addplot(buy_signals['Low']*0.99, type='scatter', 
                               markersize=100, marker='^', color='green'),
                mpf.make_addplot(sell_signals['High']*1.01, type='scatter', 
                               markersize=100, marker='v', color='red'),
            ]

            mpf.plot(df.tail(252),  # 最近一年
                    type='candle', addplot=apds, volume=True,
                    title=f'{ticker} 技术分析', 
                    savefig=f'{output_dir}/{ticker}_analysis.png',
                    figsize=(15, 10))

            # 生成统计摘要
            stats = {
                'ticker': ticker,
                'current_price': df['Close'][-1],
                'ma20': df['MA20'][-1],
                'ma60': df['MA60'][-1],
                'trend': '上涨' if df['MA20'][-1] > df['MA60'][-1] else '下跌',
                'volume_sma': df['Volume'].tail(20).mean()
            }

            print(f"{ticker} 完成: {stats}")

        except Exception as e:
            print(f"{ticker} 错误: {e}")

    print(f"报告生成完成: {output_dir}/")

# 批量处理
tickers = ['AAPL', 'MSFT', 'GOOGL', 'TSLA', 'NVDA']
generate_stock_report(tickers, '2024-01-01')

7. 实时 K 线图更新

WebSocket 实时K线

import dash
from dash import dcc, html, Input, Output
import plotly.graph_objects as go
import pandas as pd
import yfinance as yf
import time
from datetime import datetime, timedelta

app = dash.Dash(__name__)

app.layout = html.Div([
    html.H1('实时K线监控'),
    dcc.Graph(id='live-candle'),
    dcc.Interval(id='interval', interval=30*1000, n_intervals=0)  # 30秒更新
])

@app.callback(Output('live-candle', 'figure'),
              [Input('interval', 'n_intervals')])
def update_candle(n):
    # 获取最近数据
    end_time = datetime.now()
    start_time = end_time - timedelta(days=30)

    df = yf.download('AAPL', start=start_time, end=end_time, interval='1h')

    fig = go.Figure()
    fig.add_trace(go.Candlestick(
        x=df.index, open=df['Open'], high=df['High'],
        low=df['Low'], close=df['Close'],
        name='AAPL实时'
    ))

    fig.update_layout(
        title=f'AAPL 实时K线 ({datetime.now().strftime("%Y-%m-%d %H:%M")})',
        xaxis_title='时间', yaxis_title='价格'
    )

    return fig

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

8. 高级功能：Heikin-Ashi K线

平滑K线图

def heikin_ashi(df):
    """计算Heikin-Ashi K线"""
    ha = df.copy()
    ha['HA_Close'] = (df['Open'] + df['High'] + df['Low'] + df['Close']) / 4

    ha['HA_Open'] = pd.Series(index=df.index, dtype=float)
    ha['HA_Open'].iloc[0] = (df['Open'].iloc[0] + df['Close'].iloc[0]) / 2

    for i in range(1, len(ha)):
        ha['HA_Open'].iloc[i] = (ha['HA_Open'].iloc[i-1] + ha['HA_Close'].iloc[i-1]) / 2

    ha['HA_High'] = pd.concat([df['High'], ha['HA_Open'], ha['HA_Close']], axis=1).max(axis=1)
    ha['HA_Low'] = pd.concat([df['Low'], ha['HA_Open'], ha['HA_Close']], axis=1).min(axis=1)

    return ha[['HA_Open', 'HA_High', 'HA_Low', 'HA_Close', 'Volume']]

# 绘制Heikin-Ashi
df_ha = heikin_ashi(df)
mpf.plot(df_ha, type='candle', style='charles', volume=True,
         title='Heikin-Ashi K线', rename={'HA_Open':'Open', 'HA_High':'High', 
                                        'HA_Low':'Low', 'HA_Close':'Close'})

9. 最佳实践与性能优化

数据预处理

def preprocess_ohlcv(df):
    """K线数据预处理"""
    # 数据清洗
    df = df.dropna()
    df = df[df['High'] >= df['Low']]
    df = df[df['Volume'] >= 0]

    # 时间索引标准化
    df.index = pd.to_datetime(df.index)

    # 异常值检测
    price_change = df['Close'].pct_change()
    df = df[abs(price_change) < 0.5]  # 剔除日涨幅>50%的异常

    return df

# 批量预处理
df_clean = preprocess_ohlcv(df)

性能优化

# 使用缓存加速
from functools import lru_cache
import pickle

@lru_cache(maxsize=128)
def get_cached_data(ticker, start, end):
    """缓存数据获取"""
    cache_file = f"cache_{ticker}_{start}_{end}.pkl"
    try:
        with open(cache_file, 'rb') as f:
            return pickle.load(f)
    except:
        df = yf.download(ticker, start=start, end=end)
        with open(cache_file, 'wb') as f:
            pickle.dump(df, f)
        return df

10. 部署与集成

Jupyter Dashboard

from ipywidgets import interact, Dropdown, DatePicker
import ipywidgets as widgets

@interact(ticker=Dropdown(options=['AAPL', 'MSFT', 'TSLA'], value='AAPL'),
          period=widgets.IntSlider(min=30, max=365, value=90))
def plot_interactive(ticker, period):
    df = yf.download(ticker, period=f"{period}d")
    mpf.plot(df, type='candle', volume=True, style='yahoo',
             title=f'{ticker} {period}天K线')

API服务

from flask import Flask, jsonify, send_file
import io
from matplotlib.backends.backend_agg import FigureCanvasAgg as FigureCanvas

app = Flask(__name__)

@app.route('/kline/<ticker>')
def get_kline(ticker):
    df = yf.download(ticker, period='3mo')
    fig, _ = mpf.plot(df, type='candle', volume=True, 
                     returnfig=True, figsize=(12, 8))

    output = io.BytesIO()
    FigureCanvas(fig[0]).print_png(output)
    return send_file(io.BytesIO(output.getvalue()), 
                    mimetype='image/png', as_attachment=True)

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

K线图是量化分析的视觉基础，结合技术指标、交易信号和多周期分析，能显著提升策略开发效率。需要特定形态识别算法、实时数据集成或Web部署方案，请告诉我具体需求！