别再手动盯盘了！这位开发者开源的股票分析工具，让我3分钟搞定以前3小时的工作
为什么这个项目值得关注
在股市中，数据就是金矿。然而，对于普通投资者来说，获取数据、进行分析、做出决策这个过程往往耗时又费力。你是否也遇到过这些困扰？
- 想分析某只股票的历史走势，却不知道从哪里获取可靠数据
- 看着花花绿绿的K线图，却不知道哪些指标真正有价值
- 想要回测自己的交易策略，却找不到合适的工具
- 每天复盘需要手动整理大量数据，占用了宝贵的学习时间
ZhuLinsen/daily_stock_analysis 正是为解决这些痛点而生。这是一个专注于日常股票分析的 Python 工具包，它将数据获取、技术指标计算、可视化展示和策略回测等功能整合在一起，让你可以用最少的代码完成专业的股票分析。
这个项目的核心优势在于：
简洁易用：即使你是 Python 新手，也能在 5 分钟内上手运行第一个分析脚本。项目提供了清晰的 API 设计和详细的文档，让你不需要深入理解底层实现就能开始使用。
功能完整：从数据获取到技术分析，从图表绘制到策略回测，几乎涵盖了日常股票分析的所有核心需求。你不需要东拼西凑多个库，一个项目搞定全部。
代码可扩展：项目的代码结构清晰，模块化程度高。如果现有的功能不能满足你的需求，你可以轻松地基于现有代码进行扩展。
适合学习：如果你对量化投资感兴趣，这个项目是一个很好的起点。它展示了如何将金融知识转化为可运行的代码，你可以从中学习到数据处理、指标计算、策略设计等核心技能。
在接下来的教程中，我将带你从零开始，一步步掌握这个强大的股票分析工具。
环境搭建
前置要求
在开始之前，你需要确保本地环境满足以下要求：
Python 版本需要 3.8 或更高版本。你可以通过在终端中运行以下命令来检查当前 Python 版本：
python --version
如果显示的版本低于 3.8，建议先升级 Python。你可以从 Python 官方网站 (https://www.python.org) 下载最新版本，或者使用 Anaconda 这样的科学计算发行版。
创建虚拟环境（推荐）
为了避免依赖冲突，建议为项目创建独立的虚拟环境。使用 virtualenv 或 conda 都可以，这里我以 virtualenv 为例：
# 安装 virtualenv（如果还没有安装）
pip install virtualenv
# 创建新的虚拟环境
virtualenv stock_analysis_env
# 激活虚拟环境
# Windows 系统
stock_analysis_env\Scripts\activate
# macOS 和 Linux 系统
source stock_analysis_env/bin/activate
激活虚拟环境后，你的终端提示符会显示环境名称，表示你现在处于隔离的 Python 环境中。在这个环境中安装的包不会影响系统全局环境。
安装依赖包
这个项目主要依赖以下 Python 库：
- pandas：用于数据处理和分析
- numpy：用于数值计算
- matplotlib 和 plotly：用于数据可视化
- yfinance：用于获取 Yahoo Finance 的股票数据
- ta：用于计算技术指标
- akshare 或 baostock：用于获取 A 股数据（中国市场）
你可以通过项目的 requirements.txt 文件一键安装所有依赖。首先克隆项目仓库：
git clone https://github.com/ZhuLinsen/daily_stock_analysis.git
cd daily_stock_analysis
然后安装依赖：
pip install -r requirements.txt
如果项目目录中没有 requirements.txt 文件，你可以手动安装核心依赖：
pip install pandas numpy matplotlib plotly yfinance ta
对于中国A股用户，还需要安装 akshare：
pip install akshare
验证安装
安装完成后，我们来验证一下环境是否配置正确。创建一个测试脚本 check_environment.py：
"""
环境验证脚本
检查所有依赖是否正确安装
"""
import sys
def check_package(package_name, import_name=None):
"""检查单个包是否可用"""
if import_name is None:
import_name = package_name
try:
__import__(import_name)
print(f"✓ {package_name} 已安装")
return True
except ImportError:
print(f"✗ {package_name} 未安装")
return False
# 核心依赖检查
packages = [
("pandas", "pandas"),
("numpy", "numpy"),
("matplotlib", "matplotlib"),
("plotly", "plotly"),
("yfinance", "yfinance"),
("ta", "ta"),
]
print("=" * 50)
print("正在检查依赖包...")
print("=" * 50)
all_installed = all(check_package(name, imp) for name, imp in packages)
print("=" * 50)
if all_installed:
print("所有依赖包已正确安装，环境配置成功！")
print(f"Python 版本: {sys.version}")
else:
print("部分依赖包未安装，请重新安装缺失的包")
print("=" * 50)
运行这个脚本：
python check_environment.py
如果看到所有包都显示为已安装状态，说明环境配置成功，可以开始使用了。
核心功能详解
这个项目将股票分析流程分解为多个模块，每个模块负责特定的功能。让我详细介绍各个模块的设计思路和使用方法。
数据获取模块
数据是股票分析的基石。这个项目的数据获取模块支持多种数据源，可以获取全球市场的股票数据。
Yahoo Finance 数据源
对于美股、港股等国际市场，项目使用 yfinance 库获取数据。这个数据源的优势在于数据质量高、更新及时，而且完全免费。
import yfinance as yf
# 获取单只股票数据
stock = yf.Ticker("AAPL")
history = stock.history(period="1y") # 获取一年历史数据
print(history.head()) # 查看数据前几行
print(history.columns) # 查看数据列
返回的数据包含以下列：
- Open：开盘价
- High：最高价
- Low：最低价
- Close：收盘价
- Volume：成交量
- Dividends：股息
- Stock Splits：股票拆分
中国A股数据源
对于A股市场，项目集成了 akshare 库，可以获取上证、深证交易所的股票数据：
import akshare as ak
# 获取A股历史数据
df = ak.stock_zh_a_hist(symbol="000001", period="daily",
start_date="20230101", end_date="20241231")
print(df.head())
数据时间范围选择
yfinance 支持多种时间范围参数：
# 常用时间范围选项
# period="1d" - 最近1天（仅限交易日）
# period="5d" - 最近5天
# period="1mo" - 最近1个月
# period="3mo" - 最近3个月
# period="6mo" - 最近6个月
# period="1y" - 最近1年
# period="2y" - 最近2年
# period="5y" - 最近5年
# period="10y" - 最近10年
# period="ytd" - 今年至今
# period="max" - 所有可用的历史数据
# 获取苹果公司最近一年的数据
aapl = yf.Ticker("AAPL")
aapl_data = aapl.history(period="1y")
自定义时间范围
如果你需要指定具体的开始和结束日期：
# 方法1：使用 start 和 end 参数
data = yf.download("AAPL", start="2023-01-01", end="2024-01-01")
# 方法2：使用 period 和 interval 参数
# interval: 1m, 2m, 5m, 15m, 30m, 60m, 90m, 1h, 1d, 5d, 1wk, 1mo, 3mo
data = yf.download("AAPL", period="1y", interval="1d")
技术指标计算模块
技术分析是股票分析的核心。这个项目内置了常用的技术指标计算功能，基于 ta 库实现，同时也封装了一些自定义指标。
趋势指标
均线是最基础的趋势指标，包括简单移动平均线（SMA）和指数移动平均线（EMA）：
import pandas as pd
import yfinance as yf
from ta.trend import SMAIndicator, EMAIndicator
# 获取数据
data = yf.download("AAPL", period="1y")
# 计算简单移动平均线
data['SMA_20'] = SMAIndicator(data['Close'], window=20).sma_indicator()
data['SMA_50'] = SMAIndicator(data['Close'], window=50).sma_indicator()
data['SMA_200'] = SMAIndicator(data['Close'], window=200).sma_indicator()
# 计算指数移动平均线
data['EMA_12'] = EMAIndicator(data['Close'], window=12).ema_indicator()
data['EMA_26'] = EMAIndicator(data['Close'], window=26).ema_indicator()
print(data[['Close', 'SMA_20', 'SMA_50', 'EMA_12', 'EMA_26']].tail(10))
动量指标
MACD（移动平均收敛发散指标）是经典的趋势动量指标：
from ta.trend import MACD
# 计算 MACD
macd = MACD(data['Close'], window_slow=26, window_fast=12, window_sign=9)
data['MACD'] = macd.macd()
data['MACD_Signal'] = macd.macd_signal()
data['MACD_Hist'] = macd.macd_diff()
print(data[['Close', 'MACD', 'MACD_Signal', 'MACD_Hist']].tail(10))
RSI（相对强弱指标）用于判断股票的超买超卖状态：
from ta.momentum import RSIIndicator
# 计算 RSI
rsi_14 = RSIIndicator(data['Close'], window=14)
data['RSI_14'] = rsi_14.rsi()
# 也可以计算不同周期的 RSI
rsi_6 = RSIIndicator(data['Close'], window=6)
data['RSI_6'] = rsi_6.rsi()
print(data[['Close', 'RSI_14', 'RSI_6']].tail(10))
波动率指标
布林带（Bollinger Bands）是常用的波动率指标：
from ta.volatility import BollingerBands
# 计算布林带
bb = BollingerBands(data['Close'], window=20, window_dev=2)
data['BB_High'] = bb.bollinger_hband()
data['BB_Low'] = bb.bollinger_lband()
data['BB_Mid'] = bb.bollinger_mavg()
data['BB_Width'] = bb.bollinger_wband()
data['BB_Pct'] = bb.bollinger_pband()
print(data[['Close', 'BB_High', 'BB_Mid', 'BB_Low']].tail(10))
成交量指标
OBV（能量潮）结合了价格和成交量：
from ta.volume import OnBalanceVolumeIndicator
# 计算 OBV
obv = OnBalanceVolumeIndicator(data['Close'], data['Volume'])
data['OBV'] = obv.on_balance_volume()
# 计算 OBV 的移动平均
data['OBV_SMA'] = data['OBV'].rolling(window=20).mean()
print(data[['Close', 'Volume', 'OBV', 'OBV_SMA']].tail(10))
数据可视化模块
图表是理解股票走势的最佳工具。这个项目支持多种可视化方式，满足不同的分析需求。
基础K线图
import matplotlib.pyplot as plt
import mplfinance as mpf
# 准备数据（需要特定的列名格式）
plot_data = data.copy()
plot_data.columns = [col[0] if isinstance(col, tuple) else col
for col in plot_data.columns]
# 绘制K线图
mpf.plot(plot_data.tail(50), type='candle', style='charles',
title='AAPL K线图',
ylabel='价格 (USD)',
volume=True,
mav=(20, 50), # 叠加20日和50日均线
figsize=(14, 8))
技术指标叠加图
将多个指标叠加在一起显示：
import plotly.graph_objects as go
from plotly.subplots import make_subplots
# 创建子图
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=('价格与均线', '成交量', 'MACD', 'RSI'))
# K线图
fig.add_trace(go.Candlestick(x=data.index,
open=data['Open'],
high=data['High'],
low=data['Low'],
close=data['Close'],
name='K线'),
row=1, col=1)
# 均线
fig.add_trace(go.Scatter(x=data.index, y=data['SMA_20'],
mode='lines', name='SMA 20',
line=dict(color='orange')),
row=1, col=1)
fig.add_trace(go.Scatter(x=data.index, y=data['SMA_50'],
mode='lines', name='SMA 50',
line=dict(color='blue')),
row=1, col=1)
# 成交量
colors = ['green' if data['Close'].iloc[i] >= data['Open'].iloc[i]
else 'red' for i in range(len(data))]
fig.add_trace(go.Bar(x=data.index, y=data['Volume'],
marker_color=colors, name='成交量'),
row=2, col=1)
# MACD
fig.add_trace(go.Scatter(x=data.index, y=data['MACD'],
mode='lines', name='MACD',
line=dict(color='blue')),
row=3, col=1)
fig.add_trace(go.Scatter(x=data.index, y=data['MACD_Signal'],
mode='lines', name='Signal',
line=dict(color='orange')),
row=3, col=1)
fig.add_trace(go.Bar(x=data.index, y=data['MACD_Hist'],
name='Histogram'),
row=3, col=1)
# RSI
fig.add_trace(go.Scatter(x=data.index, y=data['RSI_14'],
mode='lines', name='RSI 14',
line=dict(color='purple')),
row=4, col=1)
# 添加超买超卖线
fig.add_hline(y=70, line_dash="dash", line_color="red", row=4, col=1)
fig.add_hline(y=30, line_dash="dash", line_color="green", row=4, col=1)
# 更新布局
fig.update_layout(height=1000, width=1200,
title_text='AAPL 综合技术分析',
xaxis_rangeslider_visible=False)
fig.show()
策略回测模块
回测是验证交易策略有效性的关键步骤。项目提供了简单的回测框架。
def simple_moving_average_strategy(data, short_window=20, long_window=50):
"""
简单双均线策略
当短期均线上穿长期均线时买入
当短期均线下穿长期均线时卖出
"""
signals = pd.DataFrame(index=data.index)
signals['price'] = data['Close']
signals['short_ma'] = data['Close'].rolling(window=short_window).mean()
signals['long_ma'] = data['Close'].rolling(window=long_window).mean()
# 初始化信号列
signals['signal'] = 0.0
# 生成交易信号
# 短期均线 > 长期均线 -> 1 (买入信号)
# 短期均线 < 长期均线 -> -1 (卖出信号)
signals['signal'][short_window:] = \
(signals['short_ma'][short_window:] > signals['long_ma'][short_window:]).astype(int)
# 计算持仓变化（用于标记交易点）
signals['positions'] = signals['signal'].diff()
return signals
def backtest(signals, initial_capital=10000.0):
"""
简单的回测函数
"""
# 筛选有效的信号数据
signals = signals.dropna()
# 计算每日收益率
signals['daily_returns'] = signals['price'].pct_change()
# 计算策略收益（考虑仓位）
signals['strategy_returns'] = signals['daily_returns'] * signals['signal'].shift(1)
# 计算累计收益
signals['cumulative_returns'] = (1 + signals['daily_returns']).cumprod()
signals['strategy_cumulative'] = (1 + signals['strategy_returns']).cumprod()
# 计算最终收益
final_portfolio_value = initial_capital * signals['strategy_cumulative'].iloc[-1]
# 计算统计数据
total_return = (signals['strategy_cumulative'].iloc[-1] - 1) * 100
avg_daily_return = signals['strategy_returns'].mean() * 100
std_daily_return = signals['strategy_returns'].std() * 100
# 年化收益（假设一年252个交易日）
sharpe_ratio = (avg_daily_return / std_daily_return) * (252 ** 0.5) if std_daily_return != 0 else 0
# 统计交易次数
trades = signals[signals['positions'].isin([1, -1])]
num_trades = len(trades)
print("=" * 60)
print("回测结果汇总")
print("=" * 60)
print(f"初始资金: ${initial_capital:,.2f}")
print(f"最终资金: ${final_portfolio_value:,.2f}")
print(f"总收益率: {total_return:.2f}%")
print(f"平均日收益率: {avg_daily_return:.4f}%")
print(f"收益率标准差: {std_daily_return:.4f}%")
print(f"夏普比率: {sharpe_ratio:.2f}")
print(f"交易次数: {num_trades}")
print("=" * 60)
return signals, final_portfolio_value
# 运行回测
signals = simple_moving_average_strategy(data)
results, final_value = backtest(signals)
逐步实战教程
现在让我们通过一个完整的实战案例，将所有功能串联起来。我们将分析苹果公司（AAPL）最近一年的股票数据，完成从数据获取到策略回测的全流程。
第一步：获取并整理数据
首先，创建一个完整的分析脚本 basic_analysis.py：
"""
苹果公司股票分析脚本
演示完整的数据获取、技术分析和可视化流程
"""
import pandas as pd
import numpy as np
import yfinance as yf
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore') # 忽略警告信息
def get_stock_data(ticker, days=365):
"""
获取股票历史数据
参数:
ticker: 股票代码，如 'AAPL', 'GOOGL'
days: 获取多少天的数据，默认为365天
返回:
DataFrame: 包含OHLCV数据的DataFrame
"""
print(f"正在获取 {ticker} 的历史数据...")
# 创建Ticker对象
stock = yf.Ticker(ticker)
# 获取历史数据
end_date = datetime.now()
start_date = end_date - timedelta(days=days)
data = stock.history(start=start_date, end=end_date)
print(f"成功获取 {len(data)} 个交易日的数据")
print(f"数据时间范围: {data.index[0].strftime('%Y-%m-%d')} 至 {data.index[-1].strftime('%Y-%m-%d')}")
return data
def add_technical_indicators(data):
"""
添加技术指标
参数:
data: 原始股票数据
返回:
DataFrame: 添加了技术指标的数据
"""
print("正在计算技术指标...")
df = data.copy()
# === 趋势指标 ===
# 简单移动平均线
df['SMA_20'] = df['Close'].rolling(window=20).mean()
df['SMA_50'] = df['Close'].rolling(window=50).mean()
df['SMA_200'] = df['Close'].rolling(window=200).mean()
# 指数移动平均线
df['EMA_12'] = df['Close'].ewm(span=12, adjust=False).mean()
df['EMA_26'] = df['Close'].ewm(span=26, adjust=False).mean()
# === MACD ===
df['MACD'] = df['EMA_12'] - df['EMA_26']
df['MACD_Signal'] = df['MACD'].ewm(span=9, adjust=False).mean()
df['MACD_Hist'] = df['MACD'] - df['MACD_Signal']
# === 动量指标 ===
# RSI
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
df['RSI'] = 100 - (100 / (1 + rs))
# === 波动率指标 ===
# 布林带
df['BB_Middle'] = df['Close'].rolling(window=20).mean()
df['BB_Std'] = df['Close'].rolling(window=20).std()
df['BB_Upper'] = df['BB_Middle'] + (df['BB_Std'] * 2)
df['BB_Lower'] = df['BB_Middle'] - (df['BB_Std'] * 2)
df['BB_Width'] = (df['BB_Upper'] - df['BB_Lower']) / df['BB_Middle']
df['BB_Position'] = (df['Close'] - df['BB_Lower']) / (df['BB_Upper'] - df['BB_Lower'])
# === 成交量指标 ===
# OBV
df['OBV'] = (np.sign(df['Close'].diff()) * df['Volume']).fillna(0).cumsum()
df['OBV_SMA'] = df['OBV'].rolling(window=20).mean()
# 成交量变化率
df['Volume_Change'] = df['Volume'].pct_change()
# === 市场状态 ===
# ATR (平均真实波幅)
high_low = df['High'] - df['Low']
high_close = np.abs(df['High'] - df['Close'].shift())
low_close = np.abs(df['Low'] - df['Close'].shift())
tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
df['ATR'] = tr.rolling(window=14).mean()
# 波动率百分比
df['Volatility'] = (df['ATR'] / df['Close']) * 100
print("技术指标计算完成！")
return df
def generate_signals(data):
"""
基于技术指标生成交易信号
返回:
DataFrame: 包含交易信号的数据
"""
df = data.copy()
# 初始化信号列
df['Signal'] = 0
df['Signal_Description'] = ''
# 策略1：均线金叉死叉
# 买入信号：SMA20 上穿 SMA50
# 卖出信号：SMA20 下穿 SMA50
df.loc[(df['SMA_20'] > df['SMA_50']) &
(df['SMA_20'].shift(1) <= df['SMA_50'].shift(1)), 'MA_Signal'] = 1
df.loc[(df['SMA_20'] < df['SMA_50']) &
(df['SMA_20'].shift(1) >= df['SMA_50'].shift(1)), 'MA_Signal'] = -1
# 策略2：RSI 超买超卖
# 超卖买入（RSI < 30），超买卖出（RSI > 70）
df.loc[df['RSI'] < 30, 'RSI_Signal'] = 1
df.loc[df['RSI'] > 70, 'RSI_Signal'] = -1
# 策略3：MACD 零轴穿越
df.loc[(df['MACD'] > 0) & (df['MACD'].shift(1) <= 0), 'MACD_Signal'] = 1
df.loc[(df['MACD'] < 0) & (df['MACD'].shift(1) >= 0), 'MACD_Signal'] = -1
# 综合信号
df['Combined_Score'] = (
df['MA_Signal'].fillna(0) * 0.4 +
df['RSI_Signal'].fillna(0) * 0.3 +
df['MACD_Signal'].fillna(0) * 0.3
)
# 最终交易信号
df.loc[df['Combined_Score'] > 0.5, 'Signal'] = 1 # 买入
df.loc[df['Combined_Score'] < -0.5, 'Signal'] = -1 # 卖出
return df
# 主程序
if __name__ == "__main__":
# 获取数据
ticker = "AAPL"
data = get_stock_data(ticker, days=365)
# 添加技术指标
data_with_indicators = add_technical_indicators(data)
# 生成信号
data_with_signals = generate_signals(data_with_indicators)
# 显示最新数据
print("\n" + "=" * 80)
print("最新市场数据和技术指标")
print("=" * 80)
display_cols = ['Open', 'High', 'Low', 'Close', 'Volume',
'SMA_20', 'SMA_50', 'RSI', 'MACD']
print(data_with_signals[display_cols].tail(10).to_string())
# 保存数据到CSV
output_file = f"{ticker}_analysis.csv"
data_with_signals.to_csv(output_file)
print(f"\n数据已保存到 {output_file}")
运行这个脚本：
python basic_analysis.py
你将看到类似以下的输出：
正在获取 AAPL 的历史数据...
成功获取 252 个交易日的数据
数据时间范围: 2023-01-01 至 2023-12-31
正在计算技术指标...
技术指标计算完成！
================================================================================
最新市场数据和技术指标
================================================================================
Open High Low Close Volume SMA_20 SMA_50 RSI MACD
Date
2023-12-15 185.23 186.50 184.80 186.20 52340000 185.45 183.20 58.34 2.15
2023-12-18 186.50 187.80 186.20 187.45 48560000 185.80 183.65 61.23 2.68
...
第二步：创建交互式可视化
现在让我们创建一个更高级的可视化脚本，使用 Plotly 创建交互式图表：
"""
AAPL 股票交互式可视化
使用 Plotly 创建可以缩放、平移的交互式图表
"""
import pandas as pd
import yfinance as yf
from datetime import datetime, timedelta
import plotly.graph_objects as go
from plotly.subplots import make_subplots
import warnings
warnings.filterwarnings('ignore')
def create_interactive_chart(ticker, days=365):
"""
创建交互式股票分析图表
参数:
ticker: 股票代码
days: 数据天数
"""
print(f"正在获取 {ticker} 数据并创建图表...")
# 获取数据
stock = yf.Ticker(ticker)
end_date = datetime.now()
start_date = end_date - timedelta(days=days)
data = stock.history(start=start_date, end=end_date)
# 计算技术指标
data['SMA_20'] = data['Close'].rolling(window=20).mean()
data['SMA_50'] = data['Close'].rolling(window=50).mean()
data['EMA_12'] = data['Close'].ewm(span=12, adjust=False).mean()
data['EMA_26'] = data['Close'].ewm(span=26, adjust=False).mean()
# MACD
data['MACD'] = data['EMA_12'] - data['EMA_26']
data['MACD_Signal'] = data['MACD'].ewm(span=9, adjust=False).mean()
data['MACD_Hist'] = data['MACD'] - data['MACD_Signal']
# RSI
delta = data['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
data['RSI'] = 100 - (100 / (1 + rs))
# 布林带
data['BB_Middle'] = data['Close'].rolling(window=20).mean()
data['BB_Std'] = data['Close'].rolling(window=20).std()
data['BB_Upper'] = data['BB_Middle'] + (data['BB_Std'] * 2)
data['BB_Lower'] = data['BB_Middle'] - (data['BB_Std'] * 2)
# 创建子图布局
fig = make_subplots(
rows=4, cols=1,
shared_xaxes=True,
vertical_spacing=0.03,
row_heights=[0.5, 0.15, 0.15, 0.15],
subplot_titles=('价格走势', '成交量', 'MACD', 'RSI')
)
# --- 图1：K线图和均线 ---
# K线
fig.add_trace(
go.Candlestick(
x=data.index,
open=data['Open'],
high=data['High'],
low=data['Low'],
close=data['Close'],
name='K线',
increasing_line_color='#26A69A', # 上涨颜色
decreasing_line_color='#EF5350' # 下跌颜色
),
row=1, col=1
)
# 均线
fig.add_trace(
go.Scatter(
x=data.index, y=data['SMA_20'],
mode='lines', name='SMA 20',
line=dict(color='#FF9800', width=1.5)
),
row=1, col=1
)
fig.add_trace(
go.Scatter(
x=data.index, y=data['SMA_50'],
mode='lines', name='SMA 50',
line=dict(color='#2196F3', width=1.5)
),
row=1, col=1
)
# 布林带
fig.add_trace(
go.Scatter(
x=data.index, y=data['BB_Upper'],
mode='lines', name='BB Upper',
line=dict(color='rgba(156,39,176,0.3)', width=1),
showlegend=False
),
row=1, col=1
)
fig.add_trace(
go.Scatter(
x=data.index, y=data['BB_Lower'],
mode='lines', name='BB Lower',
line=dict(color='rgba(156,39,176,0.3)', width=1),
fill='tonexty',
fillcolor='rgba(156,39,176,0.05)',
showlegend=False
),
row=1, col=1
)
# --- 图2：成交量 ---
colors = ['#26A69A' if data['Close'].iloc[i] >= data['Open'].iloc[i]
else '#EF5350' for i in range(len(data))]
fig.add_trace(
go.Bar(
x=data.index, y=data['Volume'],
marker_color=colors,
name='成交量',
opacity=0.7
),
row=2, col=1
)
# --- 图3：MACD ---
# MACD 线
fig.add_trace(
go.Scatter(
x=data.index, y=data['MACD'],
mode='lines', name='MACD',
line=dict(color='#2196F3', width=1.5)
),
row=3, col=1
)
# Signal 线
fig.add_trace(
go.Scatter(
x=data.index, y=data['MACD_Signal'],
mode='lines', name='Signal',
line=dict(color='#FF9800', width=1.5)
),
row=3, col=1
)
# MACD 柱状图
colors_hist = ['#26A69A' if val >= 0 else '#EF5350'
for val in data['MACD_Hist']]
fig.add_trace(
go.Bar(
x=data.index, y=data['MACD_Hist'],
marker_color=colors_hist,
name='MACD直方图',
opacity=0.7
),
row=3, col=1
)
# --- 图4：RSI ---
fig.add_trace(
go.Scatter(
x=data.index, y=data['RSI'],
mode='lines', name='RSI',
line=dict(color='#9C27B0', width=1.5)
),
row=4, col=1
)
# 添加超买超卖线
fig.add_hline(y=70, line_dash="dash", line_color="red",
annotation_text="超买", row=4, col=1)
fig.add_hline(y=30, line_dash="dash", line_color="green",
annotation_text="超卖", row=4, col=1)
fig.add_hline(y=50, line_dash="dot", line_color="gray", row=4, col=1)
# === 更新布局 ===
fig.update_layout(
title=dict(
text=f'{ticker} 股票技术分析',
font=dict(size=24),
x=0.5
),
height=1000,
width=1400,
template='plotly_white',
legend=dict(
orientation="h",
yanchor="bottom",
y=1.02,
xanchor="right",
x=1
),
xaxis_rangeslider_visible=False, # 隐藏顶部范围滑块
hovermode="x unified" # 统一悬停模式
)
# 更新Y轴标签
fig.update_yaxes(title_text="价格 (USD)", row=1, col=1)
fig.update_yaxes(title_text="成交量", row=2, col=1)
fig.update_yaxes(title_text="值", row=3, col=1)
fig.update_yaxes(title_text="RSI", row=4, col=1, range=[0, 100])
# 保存为HTML文件
output_file = f"{ticker}_interactive_chart.html"
fig.write_html(output_file)
print(f"交互式图表已保存到 {output_file}")
print("请在浏览器中打开该文件查看交互式图表")
# 也可以直接在浏览器中显示（如果是Jupyter环境）
# fig.show()
return fig
# 运行脚本
if __name__ == "__main__":
chart = create_interactive_chart("AAPL", days=365)
运行后，会生成一个 HTML 文件，在浏览器中打开即可看到交互式图表。你可以缩放、平移、悬停查看具体数据点。
第三步：构建完整的交易策略回测系统
现在让我们构建一个更完整的回测系统，包含资金管理、仓位控制和性能评估：
"""
完整的交易策略回测系统
包含资金管理、仓位控制和详细性能报告
"""
import pandas as pd
import numpy as np
import yfinance as yf
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import List, Tuple
import warnings
warnings.filterwarnings('ignore')
@dataclass
class Trade:
"""交易记录数据类"""
entry_date: datetime
exit_date: datetime
entry_price: float
exit_price: float
shares: int
profit: float
profit_pct: float
holding_days: int
class BacktestEngine:
"""
回测引擎
支持买入持有策略、双均线策略、RSI策略等
"""
def __init__(self, initial_capital=10000, commission=0.001):
"""
初始化回测引擎
参数:
initial_capital: 初始资金
commission: 交易佣金比例
"""
self.initial_capital = initial_capital
self.commission = commission
self.trades: List[Trade] = []
self.portfolio_value = initial_capital
self.current_position = 0 # 当前持仓股数
self.cash = initial_capital
self.position_value = 0
def reset(self):
"""重置回测状态"""
self.trades = []
self.portfolio_value = self.initial_capital
self.current_position = 0
self.cash = self.initial_capital
self.position_value = 0
def buy(self, date, price, shares=None, portion=1.0):
"""
执行买入
参数:
date: 交易日期
price: 买入价格
shares: 买入股数（如果为None，则根据portion比例计算）
portion: 资金使用比例（0-1之间）
返回:
bool: 是否成功执行
"""
if shares is None:
# 根据资金使用比例计算买入股数
available_cash = self.cash * portion
shares = int(available_cash / (price * (1 + self.commission)))
if shares <= 0:
return False
cost = shares * price * (1 + self.commission)
if cost > self.cash:
return False
self.cash -= cost
self.current_position += shares
self.position_value = self.current_position * price
return True
def sell(self, date, price, shares=None, portion=1.0):
"""
执行卖出
参数:
date: 交易日期
price: 卖出价格
shares: 卖出股数（如果为None，则根据portion比例计算）
portion: 持仓卖出比例（0-1之间）
返回:
bool: 是否成功执行
"""
if shares is None:
shares = int(self.current_position * portion)
if shares <= 0 or shares > self.current_position:
return False
proceeds = shares * price * (1 - self.commission)
# 记录交易（如果有持仓）
if self.current_position > 0:
entry_price = self.position_value / self.current_position
profit = proceeds - (shares * entry_price)
profit_pct = profit / (shares * entry_price) * 100
trade = Trade(
entry_date=date, # 简化处理
exit_date=date,
entry_price=entry_price,
exit_price=price,
shares=shares,
profit=profit,
profit_pct=profit_pct,
holding_days=0
)
self.trades.append(trade)
self.cash += proceeds
self.current_position -= shares
self.position_value = self.current_position * price
return True
def update_portfolio_value(self, price):
"""更新组合市值"""
self.position_value = self.current_position * price
self.portfolio_value = self.cash + self.position_value
def get_performance_metrics(self, trading_days_per_year=252):
"""
计算性能指标
参数:
trading_days_per_year: 每年交易日数
返回:
dict: 性能指标字典
"""
if not self.trades:
return {
'total_return': 0,
'annual_return': 0,
'win_rate': 0,
'profit_factor': 0,
'max_drawdown': 0,
'sharpe_ratio': 0,
'num_trades': 0
}
# 统计交易
profits = [t.profit for t in self.trades]
winning_trades = [p for p in profits if p > 0]
losing_trades = [p for p in profits if p < 0]
total_profit = sum(winning_trades) if winning_trades else 0
total_loss = abs(sum(losing_trades)) if losing_trades else 0
# 计算指标
total_return = (self.portfolio_value - self.initial_capital) / self.initial_capital * 100
win_rate = len(winning_trades) / len(self.trades) * 100 if self.trades else 0
profit_factor = total_profit / total_loss if total_loss > 0 else float('inf')
# 简化计算最大回撤（需要完整的权益曲线）
max_drawdown = 0 # 这里简化处理
return {
'final_value': self.portfolio_value,
'total_return': total_return,
'win_rate': win_rate,
'profit_factor': profit_factor,
'max_drawdown': max_drawdown,
'num_trades': len(self.trades),
'num_winning': len(winning_trades),
'num_losing': len(losing_trades),
'avg_profit': np.mean(winning_trades) if winning_trades else 0,
'avg_loss': np.mean(losing_trades) if losing_trades else 0
}
def dual_ma_strategy(data, short_window=20, long_window=50,
initial_capital=10000, plot=True):
"""
双均线策略回测
规则:
- 买入: 短期均线上穿长期均线
- 卖出: 短期均线下穿长期均线
"""
print("=" * 60)
print("双均线策略回测")
print("=" * 60)
print(f"短期均线: {short_window}日")
print(f"长期均线: {long_window}日")
print(f"初始资金: ${initial_capital:,.2f}")
print("=" * 60)
# 准备数据
df = data.copy()
df['SMA_Short'] = df['Close'].rolling(window=short_window).mean()
df['SMA_Long'] = df['Close'].rolling(window=long_window).mean()
# 生成信号
df['Signal'] = 0
df.loc[df['SMA_Short'] > df['SMA_Long'], 'Signal'] = 1
df.loc[df['SMA_Short'] <= df['SMA_Long'], 'Signal'] = -1
# 计算信号变化
df['Position'] = df['Signal'].diff()
# 初始化回测引擎
engine = BacktestEngine(initial_capital=initial_capital)
# 记录每日组合价值
portfolio_history = []
# 执行回测
for i, (date, row) in enumerate(df.iterrows()):
price = row['Close']
signal = row['Signal']
position_change = row['Position']
# 更新组合价值
engine.update_portfolio_value(price)
portfolio_history.append({
'date': date,
'price': price,
'position': engine.current_position,
'cash': engine.cash,
'portfolio_value': engine.portfolio_value
})
# 交易信号
if i < short_window: # 等待均线形成
continue
# 买入信号（金叉）
if position_change == 2: # 从-1变为1
engine.buy(date, price)
print(f"[买入] {date.strftime('%Y-%m-%d')} @ ${price:.2f}")
# 卖出信号（死叉）
elif position_change == -2: # 从1变为-1
if engine.current_position > 0:
engine.sell(date, price)
print(f"[卖出] {date.strftime('%Y-%m-%d')} @ ${price:.2f}")
# 平仓（回测结束时）
if engine.current_position > 0:
last_price = df.iloc[-1]['Close']
engine.sell(df.index[-1], last_price)
print(f"[最终卖出] {df.index[-1].strftime('%Y-%m-%d')} @ ${last_price:.2f}")
# 计算性能指标
metrics = engine.get_performance_metrics()
# 输出结果
print("\n" + "=" * 60)
print("回测结果")
print("=" * 60)
print(f"最终资金: ${metrics['final_value']:,.2f}")
print(f"总收益率: {metrics['total_return']:.2f}%")
print(f"交易次数: {metrics['num_trades']}")
print(f"盈利交易: {metrics['num_winning']}")
print(f"亏损交易: {metrics['num_losing']}")
print(f"胜率: {metrics['win_rate']:.2f}%")
print(f"盈亏比: {metrics['profit_factor']:.2f}")
print("=" * 60)
# 对比买入持有策略
print("\n" + "=" * 60)
print("买入持有策略对比")
print("=" * 60)
first_price = df['Close'].iloc[short_window]
last_price = df['Close'].iloc[-1]
buy_hold_return = (last_price - first_price) / first_price * 100
print(f"买入持有收益率: {buy_hold_return:.2f}%")
print(f"双均线策略收益率: {metrics['total_return']:.2f}%")
print(f"超额收益: {metrics['total_return'] - buy_hold_return:.2f}%")
print("=" * 60)
# 返回结果
result = {
'portfolio_history': pd.DataFrame(portfolio_history),
'metrics': metrics,
'trades': engine.trades,
'buy_hold_return': buy_hold_return
}
return result
def rsi_strategy(data, rsi_period=14, oversold=30, overbought=70,
initial_capital=10000):
"""
RSI 策略回测
规则:
- RSI < 30: 超卖区域，买入
- RSI > 70: 超买区域，卖出
"""
print("=" * 60)
print("RSI 策略回测")
print("=" * 60)
print(f"RSI周期: {rsi_period}")
print(f"超卖阈值: {oversold}")
print(f"超买阈值: {overbought}")
print(f"初始资金: ${initial_capital:,.2f}")
print("=" * 60)
# 准备数据
df = data.copy()
# 计算RSI
delta = df['Close'].diff()
gain = delta.where(delta > 0, 0).rolling(window=rsi_period).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=rsi_period).mean()
rs = gain / loss
df['RSI'] = 100 - (100 / (1 + rs))
# 生成交易信号
df['Signal'] = 0
df.loc[df['RSI'] < oversold, 'Signal'] = 1 # 买入
df.loc[df['RSI'] > overbought, 'Signal'] = -1 # 卖出
# 计算信号变化
df['Position'] = df['Signal'].diff()
# 初始化回测引擎
engine = BacktestEngine(initial_capital=initial_capital)
# 执行回测
for date, row in df.iterrows():
price = row['Close']
position_change = row['Position']
engine.update_portfolio_value(price)
# 买入
if position_change == 1:
engine.buy(date, price, portion=0.95) # 使用95%资金
print(f"[买入] {date.strftime('%Y-%m-%d')} @ ${price:.2f}, RSI={row['RSI']:.2f}")
# 卖出
elif position_change == -1:
if engine.current_position > 0:
engine.sell(date, price)
print(f"[卖出] {date.strftime('%Y-%m-%d')} @ ${price:.2f}, RSI={row['RSI']:.2f}")
# 平仓
if engine.current_position > 0:
last_price = df.iloc[-1]['Close']
engine.sell(df.index[-1], last_price)
# 性能指标
metrics = engine.get_performance_metrics()
print("\n" + "=" * 60)
print("回测结果")
print("=" * 60)
print(f"最终资金: ${metrics['final_value']:,.2f}")
print(f"总收益率: {metrics['total_return']:.2f}%")
print(f"交易次数: {metrics['num_trades']}")
print(f"胜率: {metrics['win_rate']:.2f}%")
print("=" * 60)
return metrics
# 主程序
if __name__ == "__main__":
# 获取数据
ticker = "AAPL"
print(f"正在获取 {ticker} 数据...")
stock = yf.Ticker(ticker)
end_date = datetime.now()
start_date = end_date - timedelta(days=365)
data = stock.history(start=start_date, end=end_date)
print(f"成功获取 {len(data)} 个交易日的数据\n")
# 运行双均线策略回测
result = dual_ma_strategy(data, short_window=20, long_window=50)
print("\n\n")
# 运行RSI策略回测
rsi_metrics = rsi_strategy(data, rsi_period=14, oversold=30, overbought=70)
运行回测脚本：
python backtest_system.py
你会看到详细的回测输出，包括每笔交易记录和最终的性能统计。
常见用例和场景
用例一：日常股票监控
如果你想每天早上快速浏览关注的股票状况，可以创建一个自动化监控脚本：
"""
每日股票监控脚本
自动获取并分析关注的股票列表
"""
import yfinance as yf
import pandas as pd
from datetime import datetime
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
def get_stock_summary(tickers):
"""
获取股票汇总信息
"""
summaries = []
for ticker in tickers:
try:
stock = yf.Ticker(ticker)
info = stock.info
# 获取最近一年的数据计算指标
hist = stock.history(period="1y")
if len(hist) < 50:
continue
# 计算均线
sma_20 = hist['Close'].tail(20).mean()
sma_50 = hist['Close'].tail(50).mean()
# 计算RSI
delta = hist['Close'].diff()
gain = delta.where(delta > 0, 0).rolling(window=14).mean().iloc[-1]
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean().iloc[-1]
rs = gain / loss if loss != 0 else 100
rsi = 100 - (100 / (1 + rs))
# 当前价格
current_price = hist['Close'].iloc[-1]
prev_close = hist['Close'].iloc[-2] if len(hist) > 1 else current_price
daily_change = (current_price - prev_close) / prev_close * 100
# 一年变化
year_start = hist['Close'].iloc[0]
year_change = (current_price - year_start) / year_start * 100
summaries.append({
'代码': ticker,
'名称': info.get('shortName', 'N/A'),
'当前价': f"${current_price:.2f}",
'日涨跌': f"{daily_change:+.2f}%",
'年涨跌': f"{year_change:+.2f}%",
'SMA20': f"${sma_20:.2f}",
'SMA50': f"${sma_50:.2f}",
'RSI': f"{rsi:.1f}",
'趋势': '多头' if sma_20 > sma_50 else '空头'
})
except Exception as e:
print(f"获取 {ticker} 数据失败: {e}")
continue
return pd.DataFrame(summaries)
def generate_alert_message(summary_df):
"""
生成告警消息
"""
message = []
message.append(f"📊 股票监控报告 - {datetime.now().strftime('%Y-%m-%d')}\n")
message.append("=" * 80)
message.append(summary_df.to_string(index=False))
message.append("\n" + "=" * 80)
# 检查异常信号
alerts = []
for _, row in summary_df.iterrows():
# RSI 超买超卖
rsi_value = float(row['RSI'])
if rsi_value > 70:
alerts.append(f"⚠️ {row['代码']} RSI超买 ({rsi_value:.1f})")
elif rsi_value < 30:
alerts.append(f"⚠️ {row['代码']} RSI超卖 ({rsi_value:.1f})")
if alerts:
message.append("\n📢 信号提醒:")
message.extend(alerts)
return "\n".join(message)
# 主程序
if __name__ == "__main__":
# 关注的股票列表
watchlist = ['AAPL', 'GOOGL', 'MSFT', 'AMZN', 'META']
# 获取汇总
summary = get_stock_summary(watchlist)
if not summary.empty:
print(generate_alert_message(summary))
# 可以保存到文件
summary.to_csv('stock_summary.csv', index=False)
print("\n汇总已保存到 stock_summary.csv")
用例二：选股策略实现
结合多个指标实现简单的选股策略：
"""
价值选股策略
基于多个财务指标筛选股票
"""
import yfinance as yf
import pandas as pd
from concurrent.futures import ThreadPoolExecutor, as_completed
def analyze_stock(ticker):
"""
分析单只股票
返回多个指标
"""
try:
stock = yf.Ticker(ticker)
info = stock.info
# 获取关键指标
pe_ratio = info.get('trailingPE', None)
pb_ratio = info.get('priceToBook', None)
roe = info.get('returnOnEquity', None)
debt_to_equity = info.get('debtToEquity', None)
dividend_yield = info.get('dividendYield', 0) * 100 if info.get('dividendYield') else 0
eps = info.get('trailingEps', None)
market_cap = info.get('marketCap', None)
# 获取历史数据
hist = stock.history(period="1y")
if len(hist) < 200:
return None
# 计算价格指标
current_price = hist['Close'].iloc[-1]
high_52w = hist['High'].max()
low_52w = hist['Low'].min()
price_to_52w_high = current_price / high_52w * 100
# 计算均线
sma_50 = hist['Close'].tail(50).mean()
sma_200 = hist['Close'].tail(200).mean() if len(hist) >= 200 else None
# 计算RSI
delta = hist['Close'].diff()
gain = delta.where(delta > 0, 0).rolling(window=14).mean().iloc[-1]
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean().iloc[-1]
rs = gain / loss if loss != 0 else 100
rsi = 100 - (100 / (1 + rs))
return {
'ticker': ticker,
'pe_ratio': pe_ratio,
'pb_ratio': pb_ratio,
'roe': roe * 100 if roe else None,
'debt_to_equity': debt_to_equity,
'dividend_yield': dividend_yield,
'eps': eps,
'market_cap': market_cap,
'price': current_price,
'52w_high': high_52w,
'52w_low': low_52w,
'price_to_52w_high': price_to_52w_high,
'sma_50': sma_50,
'sma_200': sma_200,
'rsi': rsi,
'above_sma_50': current_price > sma_50,
'above_sma_200': sma_200 and current_price > sma_200
}
except Exception as e:
print(f"分析 {ticker} 失败: {e}")
return None
def value_stock_screener(tickers):
"""
价值选股筛选器
筛选条件:
- PE < 25
- ROE > 15%
- 债务/权益比 < 100
- 价格高于52周高点50%以内
- 价格高于SMA200
- RSI < 70 (非超买)
"""
print("开始筛选股票...")
results = []
with ThreadPoolExecutor(max_workers=10) as executor:
futures = {executor.submit(analyze_stock, t): t for t in tickers}
for future in as_completed(futures):
ticker = futures[future]
try:
result = future.result()
if result:
results.append(result)
print(f"✓ {ticker} 分析完成")
except Exception as e:
print(f"✗ {ticker} 分析失败: {e}")
if not results:
print("没有找到符合条件的股票")
return pd.DataFrame()
df = pd.DataFrame(results)
# 应用筛选条件
filtered = df[
(df['pe_ratio'].notna()) & (df['pe_ratio'] < 25) &
(df['roe'].notna()) & (df['roe'] > 15) &
(df['debt_to_equity'].notna()) & (df['debt_to_equity'] < 100) &
(df['price_to_52w_high'] > 50) & # 价格高于52周低点50%以上
(df['above_sma_200'] == True) & # 价格高于SMA200
(df['rsi'] < 70) # 非超买
].copy()
# 按ROE排序
filtered = filtered.sort_values('roe', ascending=False)
return filtered
# 主程序
if __name__ == "__main__":
# 测试股票列表（实际使用时可以扩展或从文件读取）
test_tickers = [
'AAPL', 'MSFT', 'GOOGL', 'AMZN', 'META', 'NVDA', 'TSLA',
'JPM', 'JNJ', 'V', 'PG', 'UNH', 'HD', 'MA', 'DIS', 'PYPL',
'BAC', 'ADBE', 'NFLX', 'CSCO', 'INTC', 'VZ', 'T', 'PFE',
'KO', 'PEP', 'MRK', 'ABT', 'CRM', 'XOM', 'CVX', 'WMT'
]
# 运行筛选
results = value_stock_screener(test_tickers)
if not results.empty:
print("\n" + "=" * 80)
print("符合价值投资条件的股票")
print("=" * 80)
display_cols = [
'ticker', 'price', 'pe_ratio', 'roe', 'dividend_yield',
'price_to_52w_high', 'rsi'
]
print(results[display_cols].head(20).to_string(index=False))
# 保存完整结果
results.to_csv('value_stocks.csv', index=False)
print("\n完整结果已保存到 value_stocks.csv")
else:
print("没有找到符合条件的股票")
用例三：多股票对比分析
当你需要对比多只股票的走势和表现时：
"""
多股票对比分析工具
比较不同股票的表现、技术指标和相关性
"""
import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime, timedelta
def compare_stocks(tickers, period='1y'):
"""
对比多只股票
参数:
tickers: 股票代码列表
period: 分析周期
"""
data = {}
# 获取所有股票数据
for ticker in tickers:
stock = yf.Ticker(ticker)
hist = stock.history(period=period)
data[ticker] = hist['Close']
# 合并数据
df = pd.DataFrame(data)
# 归一化价格（以第一天为基准）
normalized = df / df.iloc[0] * 100
return df, normalized
def calculate_correlation(df):
"""
计算股票间的相关性
"""
return df.corr()
def calculate_performance_metrics(df):
"""
计算各股票的表现指标
"""
metrics = []
for col in df.columns:
prices = df[col]
# 基本收益
total_return = (prices.iloc[-1] - prices.iloc[0]) / prices.iloc[0] * 100
# 年化收益
days = len(prices)
annual_return = ((1 + total_return/100) ** (365/days) - 1) * 100
# 波动率
daily_returns = prices.pct_change().dropna()
volatility = daily_returns.std() * np.sqrt(252) * 100
# 夏普比率（假设无风险利率为2%）
risk_free = 0.02
sharpe = (annual_return - risk_free * 100) / volatility if volatility != 0 else 0
# 最大回撤
cumulative = (1 + daily_returns).cumprod()
peak = cumulative.expanding(min_periods=1).max()
drawdown = (cumulative - peak) / peak
max_drawdown = drawdown.min() * 100
metrics.append({
'股票代码': col,
'总收益率': f"{total_return:.2f}%",
'年化收益率': f"{annual_return:.2f}%",
'波动率': f"{volatility:.2f}%",
'夏普比率': f"{sharpe:.2f}",
'最大回撤': f"{max_drawdown:.2f}%"
})
return pd.DataFrame(metrics)
def visualize_comparison(normalized_df, normalized):
"""
可视化对比
"""
fig, axes = plt.subplots(2, 2, figsize=(16, 12))
# 图1：归一化价格走势
ax1 = axes[0, 0]
for ticker in normalized.columns:
ax1.plot(normalized.index, normalized[ticker], label=ticker, linewidth=2)
ax1.set_title('归一化价格走势 (起始=100)', fontsize=14)
ax1.set_xlabel('日期')
ax1.set_ylabel('归一化价格')
ax1.legend()
ax1.grid(True, alpha=0.3)
# 图2：收益率对比
ax2 = axes[0, 1]
returns = ((normalized.iloc[-1] - 100)).sort_values(ascending=True)
colors = ['green' if x > 0 else 'red' for x in returns.values]
returns.plot(kind='barh', ax=ax2, color=colors)
ax2.set_title('期间收益率对比', fontsize=14)
ax2.set_xlabel('收益率 (%)')
ax2.axvline(x=0, color='black', linestyle='-', linewidth=0.5)
# 图3：月度收益热力图
monthly_returns = normalized.resample('M').last().pct_change().dropna() * 100
ax3 = axes[1, 0]
im = ax3.imshow(monthly_returns.T, cmap='RdYlGn', aspect='auto')
ax3.set_yticks(range(len(monthly_returns.columns)))
ax3.set_yticklabels(monthly_returns.columns)
ax3.set_xticks(range(0, len(monthly_returns.index), 3))
ax3.set_xticklabels([d.strftime('%Y-%m') for d in monthly_returns.index[::3]], rotation=45)
ax3.set_title('月度收益率热力图', fontsize=14)
plt.colorbar(im, ax=ax3, label='收益率 (%)')
# 图4：相关性矩阵
ax4 = axes[1, 1]
corr = calculate_correlation(normalized)
im4 = ax4.imshow(corr, cmap='coolwarm', vmin=-1, vmax=1)
ax4.set_xticks(range(len(corr.columns)))
ax4.set_yticks(range(len(corr.columns)))
ax4.set_xticklabels(corr.columns)
ax4.set_yticklabels(corr.columns)
ax4.set_title('相关性矩阵', fontsize=14)
plt.colorbar(im4, ax=ax4, label='相关系数')
plt.tight_layout()
plt.savefig('stock_comparison.png', dpi=150)
plt.show()
print("对比图表已保存到 stock_comparison.png")
# 主程序
if __name__ == "__main__":
# 对比股票
compare_tickers = ['AAPL', 'MSFT', 'GOOGL', 'AMZN']
print("正在获取股票数据...")
df, normalized = compare_stocks(compare_tickers, period='1y')
print("\n" + "=" * 80)
print("股票表现对比")
print("=" * 80)
# 计算指标
metrics = calculate_performance_metrics(normalized)
print(metrics.to_string(index=False))
# 计算相关性
print("\n" + "=" * 80)
print("相关性矩阵")
print("=" * 80)
corr = calculate_correlation(normalized)
print(corr.round(3).to_string())
# 可视化
print("\n正在生成对比图表...")
visualize_comparison(df, normalized)
技巧和最佳实践
数据处理技巧
使用向量化操作代替循环
在处理大量数据时，向量化操作比循环快得多：
# ❌ 低效：使用循环计算收益率
returns = []
for i in range(1, len(prices)):
ret = (prices[i] - prices[i-1]) / prices[i-1]
returns.append(ret)
# ✅ 高效：使用 pandas 向量化操作
returns = prices.pct_change()
合理处理缺失数据
# 删除缺失值
data_clean = data.dropna()
# 用前值填充
data_filled = data.fillna(method='ffill')
# 用均值填充
data_filled = data.fillna(data.mean())
# 用插值填充（适用于时间序列）
data_interpolated = data.interpolate(method='linear')
使用缓存避免重复请求
import time
from functools import lru_cache
@lru_cache(maxsize=100)
def get_cached_data(ticker, period):
"""缓存股票数据，避免重复请求"""
print(f"从网络获取 {ticker} 数据...")
time.sleep(0.5) # 模拟网络延迟
stock = yf.Ticker(ticker)
return stock.history(period=period)
# 第一次调用会获取数据
data1 = get_cached_data("AAPL", "1y")
# 第二次调用会从缓存返回
data2 = get_cached_data("AAPL", "1y")
性能优化技巧
使用适当的数据类型
# 降低内存占用
df['Volume'] = df['Volume'].astype('int32') # 比int64节省一半内存
df['Close'] = df['Close'].astype('float32') # 比float64节省一半内存
批量获取数据
# ❌ 逐个获取（慢）
for ticker in tickers:
data = yf.download(ticker, period='1y')
# ✅ 批量获取（快）
data = yf.download(tickers, period='1y') # tickers 可以是列表
使用 Dow Jones Industrial Average 的采样技巧
# 对于长时间序列，可以降低采样频率
daily_data = stock.history(period='5y')
weekly_data = daily_data.resample('W').last() # 转为周线
monthly_data = daily_data.resample('M').last() # 转为月线
代码组织最佳实践
模块化你的代码
stock_analysis/
├── __init__.py
├── data/
│ ├── __init__.py
│ ├── fetcher.py # 数据获取
│ └── cleaner.py # 数据清洗
├── analysis/
│ ├── __init__.py
│ ├── indicators.py # 技术指标
│ ├── signals.py # 交易信号
│ └── backtest.py # 回测引擎
├── visualization/
│ ├── __init__.py
│ ├── charts.py # 图表绘制
│ └── reports.py # 报告生成
├── strategies/
│ ├── __init__.py
│ ├── ma_strategy.py
│ └── rsi_strategy.py
└── main.py
使用配置文件管理参数
# config.py
CONFIG = {
'api': {
'yfinance': {
'period': '1y',
'interval': '1d'
}
},
'indicators': {
'ma': {
'short': 20,
'medium': 50,
'long': 200
},
'rsi': {
'period': 14,
'oversold': 30,
'overbought': 70
}
},
'backtest': {
'initial_capital': 10000,
'commission': 0.001
}
}
添加适当的日志记录
import logging
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def get_data(ticker):
logger.info(f"开始获取 {ticker} 的数据")
try:
data = yf.download(ticker, period='1y')
logger.info(f"成功获取 {len(data)} 条记录")
return data
except Exception as e:
logger.error(f"获取数据失败: {e}")
raise
生产环境部署
定时任务设置
# scheduler.py
import schedule
import time
from datetime import datetime
def daily_analysis():
"""每日分析任务"""
print(f"开始执行每日分析: {datetime.now()}")
# 执行分析逻辑
# ...
print("分析完成")
def weekly_report():
"""生成周报"""
print("生成周报...")
# 生成报告逻辑
# ...
# 每天早上8点执行
schedule.every().day.at("08:00").do(daily_analysis)
# 每周一早上9点执行
schedule.every().monday.at("09:00").do(weekly_report)
while True:
schedule.run_pending()
time.sleep(60)
异常监控和处理
import traceback
import logging
from datetime import datetime
logger = logging.getLogger(__name__)
def safe_execute(func, *args, **kwargs):
"""
安全执行函数，捕获并记录异常
"""
try:
return func(*args, **kwargs)
except Exception as e:
error_msg = f"""
时间: {datetime.now()}
函数: {func.__name__}
错误: {str(e)}
堆栈:
{traceback.format_exc()}
"""
logger.error(error_msg)
# 可以添加更多处理：发送告警邮件、钉钉通知等
return None
# 使用示例
result = safe_execute(get_stock_data, "AAPL")
结论
通过这篇教程，你应该已经掌握了 ZhuLinsen/daily_stock_analysis 这个项目的核心功能和使用方法。我们从环境搭建开始，逐步深入到数据获取、技术指标计算、交易策略设计和回测验证，最后探讨了实际应用场景和最佳实践。
项目的主要价值在于：
首先，它降低了股票分析的门槛。即使你没有金融或编程背景，也能通过简洁的代码快速获取和分析股票数据。项目提供的丰富示例和清晰文档让学习曲线变得平缓。
其次，它培养量化思维。通过亲手实现各种技术指标和交易策略，你将更深入地理解市场运行的规律。比起单纯地阅读别人的分析报告，自己动手验证会让你对市场有更真实的认识。
第三，它为进阶学习奠定基础。如果你对量化投资、算法交易等更高级的领域感兴趣，这个项目是一个很好的起点。你可以在此基础上扩展更多指标、引入机器学习模型、或者对接实盘交易接口。
下一步学习建议：
如果你想继续深入学习量化投资和数据分析领域，以下资源值得关注：
- 数据科学基础：学习 pandas、numpy、scikit-learn 等数据处理和机器学习库
- 金融量化课程：Coursera、edX 上有优质的金融工程课程
- 算法交易：学习更多交易策略类型，如均值回归、统计套利、事件驱动等
- 风险控制：深入了解投资组合理论、风险度量和管理方法
- 机器学习在金融中的应用：如使用 LSTM 预测股价、使用随机森林进行选股等
GitHub 项目地址：
https://github.com/ZhuLinsen/daily_stock_analysis
如果你觉得这个项目有用，记得给作者一个 Star！也欢迎 fork 项目并贡献你自己的改进。
免责声明： 本文提供的代码和分析仅供参考学习，不构成任何投资建议。股票投资有风险，入市需谨慎。请在你充分了解相关风险后再做出任何投资决策。
评论区