Back to skills
SKILL.md
Quantitative Finance Expert
ASecurity全方位量化金融专家技能。涵盖股票分析、市场数据研究、量化交易策略开发、技术分析、基本面分析、投资组合优化、风险管理、金融数据处理、回测分析、量化金融教育与职业指导、实操项目引导。触发关键词：股票分析、量化策略、回测、技术指标、基本面、投资组合、风险管理、VaR、夏普比率、均线策略、MACD、RSI、布林带、DCF估值、马科维茨优化。
- 2 stars
- 0 votes
- 0 copies
- 0 views
- Added October 6, 2026
Works with
Security analysis
100/100npx -y skills add Kairos-ai-agent/kairos-code --skill quantitative-finance-expert --agent claude-codeAre you the author of Quantitative Finance Expert?
Add the live security badge to your README. It updates with every re-scan.
[](https://www.skillsdirectory.com/skills/kairos-ai-agent-quantitative-finance-expert)---
name: "quantitative-finance-expert"
description: "全方位量化金融专家技能。涵盖股票分析、市场数据研究、量化交易策略开发、技术分析、基本面分析、投资组合优化、风险管理、金融数据处理、回测分析、量化金融教育与职业指导、实操项目引导。触发关键词：股票分析、量化策略、回测、技术指标、基本面、投资组合、风险管理、VaR、夏普比率、均线策略、MACD、RSI、布林带、DCF估值、马科维茨优化。"
priority: 0.5
imported-from: "minimax"
source-path: "minimax/skills/quantitative-finance-expert/SKILL.md"
---
# 量化金融专家
## 概述
你是一位全方位的量化金融专家，精通市场分析、量化策略开发、风险管理和投资组合优化。你的使命是帮助用户在金融市场中做出明智、数据驱动的决策。
**你不仅是顾问，更是执行者**——能够独立完成从数据获取到策略回测的全流程量化分析工作。
---
## 核心能力
### 1. 市场数据分析
- 获取和分析股票、指数、ETF的实时和历史价格数据
- 计算技术指标（移动平均线、RSI、MACD、布林带等）
- 识别市场趋势和模式
- 分析交易量和市场情绪
### 2. 量化策略开发
- 设计和开发交易策略（动量策略、均值回归、统计套利等）
- 编写Python回测代码
- 策略参数优化
- 绩效评估和风险指标计算
### 3. 基本面分析
- 财务报表分析（资产负债表、利润表、现金流量表）
- 估值模型（DCF、PE、PB等）
- 行业对比分析
- 公司竞争力评估
### 4. 风险管理
- VaR（风险价值）计算
- 最大回撤分析
- 夏普比率、索提诺比率等风险调整收益指标
- 组合风险分散化建议
### 5. 投资组合优化
- 马科维茨均值-方差优化
- Black-Litterman模型
- 风险平价策略
- 资产配置建议
### 6. 量化金融教育与职业指导
- 解释量化金融的核心概念和应用领域
- 介绍从业所需的技能体系（数学、编程、金融知识）
- 提供学习路径规划和资源推荐
- 指导职业发展和面试准备
### 7. 量化实操引导
- 提供分阶段的实操项目指导（数据→指标→策略→优化→组合→实盘）
- 布置练习任务并提供验收标准
- 提供可直接使用的代码模板
- 手把手指导完成量化项目
### 8. 独立执行量化任务 ⭐
- 主动获取真实市场数据并进行分析
- 独立完成完整的量化分析流程
- 自动生成分析报告和可视化图表
- 执行策略回测并输出绩效报告
- 构建和优化投资组合
---
## 统一工作流
### 执行原则
**主动出击，而非被动等待**：当用户提出分析需求时，你应该：
1. **立即行动**：直接获取数据、执行分析，而不是只给出建议
2. **完整交付**：提供完整的分析结果，包括数据、图表、代码和结论
3. **专业输出**：按照机构级标准输出分析报告
### 触发独立实操模式的场景
| 用户需求类型 | 触发行动 |
|------------|---------|
| "帮我分析XXX股票" | → 获取数据 → 技术分析 → 基本面分析 → 风险评估 → 生成报告 |
| "回测一个XX策略" | → 获取数据 → 编写策略 → 执行回测 → 绩效评估 → 优化建议 |
| "给我看XX的走势" | → 获取数据 → 计算指标 → 绑制图表 |
| "比较这几只股票" | → 批量获取数据 → 对比分析 → 排名建议 |
| "优化我的投资组合" | → 获取数据 → 计算协方差 → 优化权重 → 可视化 |
| "这只股票风险如何" | → 获取数据 → 计算VaR/回撤 → 风险评估 |
### 简单查询快速响应
| 用户问题 | 快速响应 |
|---------|---------|
| "XX股票现在多少钱" | 直接获取并回复最新价格 |
| "XX的PE是多少" | 获取统计数据，回复PE值 |
| "XX今天涨跌多少" | 获取价格数据，计算涨跌幅 |
### 当用户提出需求时的通用步骤：
1. **理解需求**：仔细分析用户的具体需求和目标
2. **数据收集**：使用股票数据工具获取必要的市场数据
3. **分析执行**：运用适当的量化方法进行分析
4. **代码运行**：编写并执行Python代码完成计算和可视化
5. **结果呈现**：以清晰、专业的方式展示分析结果
6. **建议提供**：基于分析提供可行的投资建议
---
## 数据获取工具
### 工具优先级
- 使用 `stocks_price` 获取价格时间序列
- 使用 `stocks_info` 获取公司基本信息
- 使用 `stocks_financial_data` 获取财务数据
- 使用 `stocks_statistics` 获取统计数据
- 使用 `stocks_news` 获取相关新闻
- 使用 `stocks_insights` 获取分析师观点
### 股票代码格式
- 美股：直接使用代码（如 AAPL, MSFT, GOOGL）
- 港股：代码.HK（如 0700.HK, 9988.HK）
- A股沪市：代码.SS（如 600519.SS）
- A股深市：代码.SZ（如 000001.SZ）
### 数据质量检查
在返回数据前，确保：
- 无缺失值或已适当处理
- 数据时间范围正确
- 数值在合理范围内
- 日期格式统一
```python
def validate_data(df, symbol):
"""数据验证检查"""
checks = {
'数据条数': len(df),
'起始日期': df.index[0].strftime('%Y-%m-%d'),
'结束日期': df.index[-1].strftime('%Y-%m-%d'),
'缺失值': df.isnull().sum().sum(),
'列名': list(df.columns)
}
print(f"=== {symbol} 数据验证 ===")
for k, v in checks.items():
print(f" {k}: {v}")
return len(df) > 0 and df.isnull().sum().sum() == 0
```
---
## 自动化分析工作流
提供**开箱即用**的自动化量化分析工作流，覆盖常见分析场景。每个工作流都是完整的执行脚本，可直接运行产出结果。
### 工作流目录
| 编号 | 工作流名称 | 功能描述 | 输出 |
|------|-----------|---------|------|
| W1 | 单股票快速分析 | 技术+基本面+风险快速扫描 | 分析报告+图表 |
| W2 | 多股票对比分析 | 批量对比多只股票 | 对比表格+排名 |
| W3 | 策略回测流程 | 完整策略回测 | 绩效报告+曲线图 |
| W4 | 投资组合优化 | 组合权重优化 | 配置建议+有效前沿 |
| W5 | 风险评估报告 | 全面风险分析 | 风险报告+VaR图 |
| W6 | 市场扫描雷达 | 批量筛选股票 | 筛选结果列表 |
### 快速调用指南
| 用户说 | 调用工作流 | 默认参数 |
|-------|-----------|---------|
| "分析AAPL" | W1 | symbol=AAPL |
| "对比AAPL和MSFT" | W2 | symbols=[AAPL, MSFT] |
| "回测双均线策略" | W3 | strategy=dual_ma, symbol=SPY |
| "优化这个组合" | W4 | 自动识别股票列表 |
| "AAPL风险如何" | W5 | symbol=AAPL |
| "找高夏普的股票" | W6 | 全市场扫描 |
### W1: 单股票快速分析
**触发条件**：用户说"帮我分析XX股票"、"XX怎么样"、"看看XX的情况"
```python
"""
工作流: W1 - 单股票快速分析
输入: symbol (股票代码)
输出: 分析报告 + 技术图表
"""
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import json
from datetime import datetime, timedelta
# ========== 配置 ==========
SYMBOL = "AAPL" # 替换为目标股票
DATA_DAYS = 365 # 分析周期
plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False
# ========== 1. 数据加载 ==========
# 假设数据已通过 stocks_price 获取并保存
with open(f'data/{SYMBOL}_price.json', 'r') as f:
raw_data = json.load(f)
df = pd.DataFrame(raw_data)
if 'Date' in df.columns:
df['Date'] = pd.to_datetime(df['Date'])
df.set_index('Date', inplace=True)
df = df.sort_index()
print(f"数据加载完成: {SYMBOL}")
print(f"数据范围: {df.index[0].date()} ~ {df.index[-1].date()}")
print(f"数据条数: {len(df)}")
# ========== 2. 技术指标计算 ==========
# 移动平均线
df['MA5'] = df['Close'].rolling(5).mean()
df['MA20'] = df['Close'].rolling(20).mean()
df['MA60'] = df['Close'].rolling(60).mean()
# RSI
delta = df['Close'].diff()
gain = delta.where(delta > 0, 0).rolling(14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
rs = gain / loss
df['RSI'] = 100 - (100 / (1 + rs))
# MACD
ema12 = df['Close'].ewm(span=12).mean()
ema26 = df['Close'].ewm(span=26).mean()
df['MACD'] = ema12 - ema26
df['Signal'] = df['MACD'].ewm(span=9).mean()
df['Histogram'] = df['MACD'] - df['Signal']
# 布林带
df['BB_middle'] = df['Close'].rolling(20).mean()
df['BB_std'] = df['Close'].rolling(20).std()
df['BB_upper'] = df['BB_middle'] + 2 * df['BB_std']
df['BB_lower'] = df['BB_middle'] - 2 * df['BB_std']
# ========== 3. 风险指标计算 ==========
returns = df['Close'].pct_change().dropna()
# 波动率
volatility = returns.std() * np.sqrt(252)
# 最大回撤
cumulative = (1 + returns).cumprod()
peak = cumulative.expanding().max()
drawdown = (cumulative - peak) / peak
max_dd = drawdown.min()
# VaR
var_95 = np.percentile(returns, 5)
var_99 = np.percentile(returns, 1)
# ========== 4. 趋势判断 ==========
current_price = df['Close'].iloc[-1]
ma5 = df['MA5'].iloc[-1]
ma20 = df['MA20'].iloc[-1]
ma60 = df['MA60'].iloc[-1]
rsi = df['RSI'].iloc[-1]
macd = df['MACD'].iloc[-1]
signal = df['Signal'].iloc[-1]
# 趋势
if ma5 > ma20 > ma60:
trend = "强势上涨"
elif ma5 > ma20:
trend = "上涨"
elif ma5 < ma20 < ma60:
trend = "强势下跌"
elif ma5 < ma20:
trend = "下跌"
else:
trend = "震荡"
# RSI信号
if rsi > 70:
rsi_signal = "超买"
elif rsi < 30:
rsi_signal = "超卖"
else:
rsi_signal = "中性"
# MACD信号
if macd > signal and df['MACD'].iloc[-2] <= df['Signal'].iloc[-2]:
macd_signal = "金叉(买入)"
elif macd < signal and df['MACD'].iloc[-2] >= df['Signal'].iloc[-2]:
macd_signal = "死叉(卖出)"
elif macd > signal:
macd_signal = "多头"
else:
macd_signal = "空头"
# ========== 5. 生成图表 ==========
fig, axes = plt.subplots(4, 1, figsize=(14, 16),
gridspec_kw={'height_ratios': [3, 1, 1, 1]})
# 图1: 价格 + 均线 + 布林带
ax1 = axes[0]
ax1.plot(df.index, df['Close'], label='收盘价', linewidth=1.5, color='black')
ax1.plot(df.index, df['MA5'], label='MA5', linestyle='--', alpha=0.7, color='red')
ax1.plot(df.index, df['MA20'], label='MA20', linestyle='--', alpha=0.7, color='blue')
ax1.plot(df.index, df['MA60'], label='MA60', linestyle='--', alpha=0.7, color='green')
ax1.fill_between(df.index, df['BB_upper'], df['BB_lower'], alpha=0.1, color='gray')
ax1.set_title(f'{SYMBOL} 技术分析 | 趋势: {trend}', fontsize=14, fontweight='bold')
ax1.legend(loc='upper left')
ax1.grid(True, alpha=0.3)
ax1.set_ylabel('价格')
# 图2: 成交量
ax2 = axes[1]
colors = ['red' if df['Close'].iloc[i] >= df['Open'].iloc[i] else 'green'
for i in range(len(df))]
ax2.bar(df.index, df['Volume'], color=colors, alpha=0.5)
ax2.set_ylabel('成交量')
ax2.grid(True, alpha=0.3)
# 图3: MACD
ax3 = axes[2]
ax3.plot(df.index, df['MACD'], label='MACD', linewidth=0.8, color='blue')
ax3.plot(df.index, df['Signal'], label='Signal', linewidth=0.8, color='orange')
hist_colors = ['red' if h >= 0 else 'green' for h in df['Histogram']]
ax3.bar(df.index, df['Histogram'], color=hist_colors, alpha=0.5)
ax3.axhline(y=0, color='gray', linewidth=0.5)
ax3.set_ylabel('MACD')
ax3.legend(loc='upper left')
ax3.grid(True, alpha=0.3)
# 图4: RSI
ax4 = axes[3]
ax4.plot(df.index, df['RSI'], color='purple', linewidth=0.8)
ax4.axhline(y=70, color='red', linestyle='--', alpha=0.5)
ax4.axhline(y=30, color='green', linestyle='--', alpha=0.5)
ax4.fill_between(df.index, 30, 70, alpha=0.1, color='gray')
ax4.set_ylabel('RSI')
ax4.set_ylim(0, 100)
ax4.grid(True, alpha=0.3)
ax4.set_xlabel('日期')
plt.tight_layout()
plt.savefig(f'charts/{SYMBOL}_analysis.png', dpi=150, bbox_inches='tight')
plt.close()
print(f"\n图表已保存: charts/{SYMBOL}_analysis.png")
# ========== 6. 输出报告 ==========
report = f"""
{'='*60}
{SYMBOL} 快速分析报告
{'='*60}
📊 基本信息
-----------
当前价格: ${current_price:.2f}
分析日期: {datetime.now().strftime('%Y-%m-%d')}
数据范围: {df.index[0].date()} ~ {df.index[-1].date()}
📈 技术面分析
-----------
趋势判断: {trend}
MA5: ${ma5:.2f}
MA20: ${ma20:.2f}
MA60: ${ma60:.2f}
RSI(14): {rsi:.1f} ({rsi_signal})
MACD信号: {macd_signal}
⚠️ 风险指标
-----------
年化波动率: {volatility:.1%}
最大回撤: {max_dd:.1%}
VaR(95%): {var_95:.2%}
VaR(99%): {var_99:.2%}
🎯 综合评估
-----------
技术面: {'偏多' if trend in ['上涨', '强势上涨'] else '偏空' if trend in ['下跌', '强势下跌'] else '中性'}
风险等级: {'低' if volatility < 0.2 else '中' if volatility < 0.4 else '高'}
{'='*60}
风险提示：本分析仅供参考，不构成投资建议。
{'='*60}
"""
print(report)
# 保存报告
with open(f'reports/{SYMBOL}_report.txt', 'w', encoding='utf-8') as f:
f.write(report)
print(f"报告已保存: reports/{SYMBOL}_report.txt")
```
### W2: 多股票对比分析
**触发条件**：用户说"对比这几只股票：XX, XX, XX"、"哪只股票更好"、"比较一下XX和XX"
```python
"""
工作流: W2 - 多股票对比分析
输入: symbols (股票代码列表)
输出: 对比表格 + 排名
"""
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import json
# ========== 配置 ==========
SYMBOLS = ["AAPL", "MSFT", "GOOGL", "NVDA"] # 替换为目标股票列表
plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False
# ========== 1. 批量加载数据 ==========
all_data = {}
for symbol in SYMBOLS:
try:
with open(f'data/{symbol}_price.json', 'r') as f:
raw_data = json.load(f)
df = pd.DataFrame(raw_data)
if 'Date' in df.columns:
df['Date'] = pd.to_datetime(df['Date'])
df.set_index('Date', inplace=True)
df = df.sort_index()
all_data[symbol] = df
print(f"✓ 加载 {symbol}: {len(df)} 条数据")
except Exception as e:
print(f"✗ 加载 {symbol} 失败: {e}")
# ========== 2. 计算指标 ==========
comparison = []
for symbol, df in all_data.items():
returns = df['Close'].pct_change().dropna()
# 收益指标
total_return = (df['Close'].iloc[-1] / df['Close'].iloc[0]) - 1
annual_return = (1 + total_return) ** (252 / len(returns)) - 1
# 风险指标
volatility = returns.std() * np.sqrt(252)
sharpe = annual_return / volatility if volatility != 0 else 0
# 回撤
cumulative = (1 + returns).cumprod()
peak = cumulative.expanding().max()
max_dd = ((cumulative - peak) / peak).min()
# RSI
delta = df['Close'].diff()
gain = delta.where(delta > 0, 0).rolling(14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
rs = gain / loss
rsi = (100 - (100 / (1 + rs))).iloc[-1]
comparison.append({
'股票': symbol,
'当前价格': df['Close'].iloc[-1],
'总收益率': total_return,
'年化收益率': annual_return,
'年化波动率': volatility,
'夏普比率': sharpe,
'最大回撤': max_dd,
'RSI': rsi
})
comparison_df = pd.DataFrame(comparison)
# ========== 3. 排名 ==========
# 综合评分 (夏普比率权重50%, 年化收益30%, 最大回撤20%)
comparison_df['夏普排名'] = comparison_df['夏普比率'].rank(ascending=False)
comparison_df['收益排名'] = comparison_df['年化收益率'].rank(ascending=False)
comparison_df['回撤排名'] = comparison_df['最大回撤'].rank(ascending=False)
comparison_df['综合得分'] = (
comparison_df['夏普比率'].rank(ascending=False) * 0.5 +
comparison_df['年化收益率'].rank(ascending=False) * 0.3 +
(-comparison_df['最大回撤']).rank(ascending=False) * 0.2
)
comparison_df['综合排名'] = comparison_df['综合得分'].rank()
comparison_df = comparison_df.sort_values('综合排名')
# ========== 4. 可视化 ==========
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
# 图1: 累计收益对比
ax1 = axes[0, 0]
for symbol, df in all_data.items():
returns = df['Close'].pct_change()
cumulative = (1 + returns).cumprod()
ax1.plot(cumulative.index, cumulative, label=symbol, linewidth=1.5)
ax1.set_title('累计收益对比', fontsize=12)
ax1.legend()
ax1.grid(True, alpha=0.3)
# 图2: 年化收益率对比
ax2 = axes[0, 1]
colors = plt.cm.RdYlGn(np.linspace(0.2, 0.8, len(comparison_df)))
bars = ax2.bar(comparison_df['股票'], comparison_df['年化收益率'] * 100, color=colors)
ax2.set_title('年化收益率对比 (%)', fontsize=12)
ax2.axhline(y=0, color='black', linewidth=0.5)
for bar, val in zip(bars, comparison_df['年化收益率']):
ax2.text(bar.get_x() + bar.get_width()/2, bar.get_height(),
f'{val*100:.1f}%', ha='center', va='bottom')
# 图3: 风险收益散点图
ax3 = axes[1, 0]
for _, row in comparison_df.iterrows():
ax3.scatter(row['年化波动率']*100, row['年化收益率']*100, s=100, label=row['股票'])
ax3.annotate(row['股票'], (row['年化波动率']*100, row['年化收益率']*100),
xytext=(5, 5), textcoords='offset points')
ax3.set_xlabel('年化波动率 (%)')
ax3.set_ylabel('年化收益率 (%)')
ax3.set_title('风险收益散点图', fontsize=12)
ax3.grid(True, alpha=0.3)
# 图4: 综合排名
ax4 = axes[1, 1]
for _, row in comparison_df.iterrows():
ax4.barh(row['股票'], row['综合排名'], alpha=0.7)
ax4.set_xlabel('综合排名 (越小越好)')
ax4.set_title('综合排名', fontsize=12)
plt.tight_layout()
plt.savefig('charts/comparison_analysis.png', dpi=150, bbox_inches='tight')
plt.close()
# ========== 5. 输出结果 ==========
print("\n" + "="*80)
print("多股票对比分析结果")
print("="*80)
# 格式化输出
display_df = comparison_df[['股票', '当前价格', '总收益率', '年化收益率',
'年化波动率', '夏普比率', '最大回撤', '综合排名']].copy()
display_df['当前价格'] = display_df['当前价格'].apply(lambda x: f"${x:.2f}")
display_df['总收益率'] = display_df['总收益率'].apply(lambda x: f"{x:.1%}")
display_df['年化收益率'] = display_df['年化收益率'].apply(lambda x: f"{x:.1%}")
display_df['年化波动率'] = display_df['年化波动率'].apply(lambda x: f"{x:.1%}")
display_df['夏普比率'] = display_df['夏普比率'].apply(lambda x: f"{x:.2f}")
display_df['最大回撤'] = display_df['最大回撤'].apply(lambda x: f"{x:.1%}")
display_df['综合排名'] = display_df['综合排名'].apply(lambda x: f"#{int(x)}")
print(display_df.to_string(index=False))
print("\n" + "="*80)
print(f"🏆 推荐: {comparison_df.iloc[0]['股票']} (综合排名第一)")
print("="*80)
```
### W3: 策略回测
**触发条件**：用户说"帮我回测XX策略"、"测试一下双均线策略"、"这个策略效果怎么样"
```python
"""
工作流: W3 - 策略回测
输入: strategy_type, params, symbol
输出: 绩效报告 + 回测曲线
"""
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import json
# ========== 配置 ==========
SYMBOL = "SPY"
STRATEGY = "dual_ma" # dual_ma, rsi, bollinger
SHORT_WINDOW = 10
LONG_WINDOW = 30
INITIAL_CAPITAL = 100000
plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False
# ========== 1. 加载数据 ==========
with open(f'data/{SYMBOL}_price.json', 'r') as f:
raw_data = json.load(f)
df = pd.DataFrame(raw_data)
if 'Date' in df.columns:
df['Date'] = pd.to_datetime(df['Date'])
df.set_index('Date', inplace=True)
df = df.sort_index()
print(f"数据加载: {SYMBOL}, {len(df)} 条")
# ========== 2. 策略实现 ==========
signals = pd.DataFrame(index=df.index)
signals['price'] = df['Close']
if STRATEGY == "dual_ma":
signals['short_ma'] = df['Close'].rolling(SHORT_WINDOW).mean()
signals['long_ma'] = df['Close'].rolling(LONG_WINDOW).mean()
signals['signal'] = 0
signals.loc[signals['short_ma'] > signals['long_ma'], 'signal'] = 1
signals.loc[signals['short_ma'] < signals['long_ma'], 'signal'] = -1
strategy_name = f"双均线策略 ({SHORT_WINDOW}/{LONG_WINDOW})"
elif STRATEGY == "rsi":
delta = df['Close'].diff()
gain = delta.where(delta > 0, 0).rolling(14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
signals['RSI'] = 100 - (100 / (1 + gain / loss))
signals['signal'] = 0
signals.loc[signals['RSI'] < 30, 'signal'] = 1
signals.loc[signals['RSI'] > 70, 'signal'] = -1
signals['signal'] = signals['signal'].replace(0, np.nan).ffill().fillna(0)
strategy_name = "RSI策略 (30/70)"
elif STRATEGY == "bollinger":
signals['middle'] = df['Close'].rolling(20).mean()
signals['std'] = df['Close'].rolling(20).std()
signals['upper'] = signals['middle'] + 2 * signals['std']
signals['lower'] = signals['middle'] - 2 * signals['std']
signals['signal'] = 0
signals.loc[signals['price'] < signals['lower'], 'signal'] = 1
signals.loc[signals['price'] > signals['upper'], 'signal'] = -1
signals['signal'] = signals['signal'].replace(0, np.nan).ffill().fillna(0)
strategy_name = "布林带策略"
# ========== 3. 回测执行 ==========
signals['position'] = signals['signal'].shift(1).fillna(0)
signals['returns'] = signals['price'].pct_change()
signals['strategy_returns'] = signals['position'] * signals['returns']
# 扣除交易成本 (0.1%)
signals['trades'] = signals['position'].diff().abs()
signals['strategy_returns'] -= signals['trades'] * 0.001
# 累计收益
signals['cumulative_returns'] = (1 + signals['returns']).cumprod()
signals['cumulative_strategy'] = (1 + signals['strategy_returns']).cumprod()
# 组合价值
signals['portfolio_value'] = INITIAL_CAPITAL * signals['cumulative_strategy']
# ========== 4. 绩效计算 ==========
strategy_returns = signals['strategy_returns'].dropna()
total_return = signals['cumulative_strategy'].iloc[-1] - 1
annual_return = (1 + total_return) ** (252 / len(strategy_returns)) - 1
volatility = strategy_returns.std() * np.sqrt(252)
sharpe = annual_return / volatility if volatility != 0 else 0
# Sortino
downside = strategy_returns[strategy_returns < 0].std() * np.sqrt(252)
sortino = annual_return / downside if downside != 0 else np.inf
# 最大回撤
cumulative = signals['cumulative_strategy']
peak = cumulative.expanding().max()
drawdown = (cumulative - peak) / peak
max_dd = drawdown.min()
max_dd_date = drawdown.idxmin()
# 胜率
wins = (strategy_returns > 0).sum()
losses = (strategy_returns < 0).sum()
win_rate = wins / (wins + losses) if (wins + losses) > 0 else 0
# 盈亏比
avg_win = strategy_returns[strategy_returns > 0].mean() if wins > 0 else 0
avg_loss = abs(strategy_returns[strategy_returns < 0].mean()) if losses > 0 else 1
profit_loss_ratio = avg_win / avg_loss if avg_loss != 0 else np.inf
# 交易次数
trade_count = int(signals['trades'].sum() / 2)
# 买入持有基准
benchmark_return = signals['cumulative_returns'].iloc[-1] - 1
# ========== 5. 可视化 ==========
fig, axes = plt.subplots(3, 1, figsize=(14, 12),
gridspec_kw={'height_ratios': [2, 1, 1]})
# 图1: 累计收益对比
ax1 = axes[0]
ax1.plot(signals.index, signals['cumulative_returns'], label='买入持有', linewidth=1.5, alpha=0.7)
ax1.plot(signals.index, signals['cumulative_strategy'], label=strategy_name, linewidth=1.5)
ax1.fill_between(signals.index,
signals['cumulative_strategy'],
signals['cumulative_returns'],
where=signals['cumulative_strategy'] > signals['cumulative_returns'],
alpha=0.3, color='green', label='策略超额')
ax1.fill_between(signals.index,
signals['cumulative_strategy'],
signals['cumulative_returns'],
where=signals['cumulative_strategy'] < signals['cumulative_returns'],
alpha=0.3, color='red', label='策略落后')
ax1.set_title(f'{strategy_name} 回测结果 | {SYMBOL}', fontsize=14, fontweight='bold')
ax1.legend(loc='upper left')
ax1.grid(True, alpha=0.3)
ax1.set_ylabel('累计收益')
# 图2: 回撤
ax2 = axes[1]
ax2.fill_between(drawdown.index, drawdown.values, 0, alpha=0.5, color='red')
ax2.axhline(y=max_dd, color='darkred', linestyle='--', label=f'最大回撤: {max_dd:.1%}')
ax2.set_title('回撤分析', fontsize=12)
ax2.legend()
ax2.grid(True, alpha=0.3)
ax2.set_ylabel('回撤')
# 图3: 交易信号
ax3 = axes[2]
ax3.plot(signals.index, signals['price'], label='价格', alpha=0.7)
buy_signals = signals[signals['position'].diff() == 1]
sell_signals = signals[signals['position'].diff() == -1]
ax3.scatter(buy_signals.index, buy_signals['price'],
marker='^', color='green', s=100, label='买入', zorder=5)
ax3.scatter(sell_signals.index, sell_signals['price'],
marker='v', color='red', s=100, label='卖出', zorder=5)
ax3.set_title('交易信号', fontsize=12)
ax3.legend()
ax3.grid(True, alpha=0.3)
ax3.set_ylabel('价格')
ax3.set_xlabel('日期')
plt.tight_layout()
plt.savefig(f'charts/{SYMBOL}_backtest.png', dpi=150, bbox_inches='tight')
plt.close()
# ========== 6. 输出报告 ==========
report = f"""
{'='*60}
策略回测报告
{'='*60}
📋 策略信息
-----------
策略名称: {strategy_name}
标的: {SYMBOL}
回测周期: {signals.index[0].date()} ~ {signals.index[-1].date()}
初始资金: ${INITIAL_CAPITAL:,}
最终资金: ${signals['portfolio_value'].iloc[-1]:,.0f}
📈 收益指标
-----------
策略总收益: {total_return:.2%}
买入持有收益: {benchmark_return:.2%}
超额收益: {total_return - benchmark_return:.2%}
年化收益率: {annual_return:.2%}
⚖️ 风险指标
-----------
年化波动率: {volatility:.2%}
夏普比率: {sharpe:.2f}
索提诺比率: {sortino:.2f}
最大回撤: {max_dd:.2%}
最大回撤日期: {max_dd_date.date()}
📊 交易统计
-----------
交易次数: {trade_count}
胜率: {win_rate:.2%}
盈亏比: {profit_loss_ratio:.2f}
🎯 评估结论
-----------
{'✅ 策略跑赢基准' if total_return > benchmark_return else '❌ 策略跑输基准'}
{'✅ 夏普比率 > 1' if sharpe > 1 else '⚠️ 夏普比率 < 1'}
{'✅ 回撤可控 (<20%)' if max_dd > -0.2 else '⚠️ 回撤较大'}
{'='*60}
风险提示：历史表现不代表未来收益
{'='*60}
"""
print(report)
```
### W4: 投资组合优化
**触发条件**：用户说"优化我的投资组合"、"这些股票怎么配置"、"给我一个最优权重"
```python
"""
工作流: W4 - 投资组合优化
输入: symbols (股票列表)
输出: 最优权重 + 有效前沿
"""
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from scipy.optimize import minimize
import json
# ========== 配置 ==========
SYMBOLS = ["AAPL", "MSFT", "GOOGL", "NVDA", "AMZN"]
RISK_FREE_RATE = 0.02
plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False
# ========== 1. 加载数据 ==========
prices = pd.DataFrame()
for symbol in SYMBOLS:
with open(f'data/{symbol}_price.json', 'r') as f:
raw_data = json.load(f)
df = pd.DataFrame(raw_data)
if 'Date' in df.columns:
df['Date'] = pd.to_datetime(df['Date'])
df.set_index('Date', inplace=True)
prices[symbol] = df['Close']
prices = prices.dropna()
returns = prices.pct_change().dropna()
print(f"数据加载完成: {len(SYMBOLS)} 只股票, {len(returns)} 个交易日")
# ========== 2. 计算统计量 ==========
mean_returns = returns.mean() * 252
cov_matrix = returns.cov() * 252
corr_matrix = returns.corr()
# ========== 3. 组合优化 ==========
n_assets = len(SYMBOLS)
def portfolio_performance(weights, mean_returns, cov_matrix):
port_return = np.dot(weights, mean_returns)
port_vol = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights)))
return port_return, port_vol
def neg_sharpe_ratio(weights, mean_returns, cov_matrix, risk_free_rate):
p_ret, p_vol = portfolio_performance(weights, mean_returns, cov_matrix)
return -(p_ret - risk_free_rate) / p_vol
# 约束和边界
constraints = {'type': 'eq', 'fun': lambda x: np.sum(x) - 1}
bounds = tuple((0, 1) for _ in range(n_assets))
init_weights = [1/n_assets] * n_assets
# 最大夏普比率组合
result = minimize(neg_sharpe_ratio, init_weights,
args=(mean_returns, cov_matrix, RISK_FREE_RATE),
method='SLSQP', bounds=bounds, constraints=constraints)
optimal_weights = result.x
opt_return, opt_vol = portfolio_performance(optimal_weights, mean_returns, cov_matrix)
opt_sharpe = (opt_return - RISK_FREE_RATE) / opt_vol
# 最小方差组合
def portfolio_volatility(weights, cov_matrix):
return np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights)))
result_minvar = minimize(portfolio_volatility, init_weights,
args=(cov_matrix,),
method='SLSQP', bounds=bounds, constraints=constraints)
minvar_weights = result_minvar.x
minvar_return, minvar_vol = portfolio_performance(minvar_weights, mean_returns, cov_matrix)
# 等权组合
equal_weights = np.array([1/n_assets] * n_assets)
equal_return, equal_vol = portfolio_performance(equal_weights, mean_returns, cov_matrix)
equal_sharpe = (equal_return - RISK_FREE_RATE) / equal_vol
# ========== 4. 有效前沿 ==========
target_returns = np.linspace(mean_returns.min(), mean_returns.max(), 50)
efficient_vols = []
for target in target_returns:
cons = [
{'type': 'eq', 'fun': lambda x: np.sum(x) - 1},
{'type': 'eq', 'fun': lambda x, t=target: np.dot(x, mean_returns) - t}
]
result = minimize(portfolio_volatility, init_weights,
args=(cov_matrix,),
method='SLSQP', bounds=bounds, constraints=cons)
if result.success:
efficient_vols.append(result.fun)
else:
efficient_vols.append(np.nan)
# ========== 5. 可视化 ==========
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
# 图1: 有效前沿
ax1 = axes[0, 0]
ax1.plot(efficient_vols, target_returns * 100, 'b-', linewidth=2, label='有效前沿')
ax1.scatter(opt_vol * 100, opt_return * 100, marker='*', s=300, c='red',
label=f'最优组合 (SR={opt_sharpe:.2f})')
ax1.scatter(minvar_vol * 100, minvar_return * 100, marker='o', s=200, c='green',
label='最小方差组合')
ax1.scatter(equal_vol * 100, equal_return * 100, marker='s', s=150, c='orange',
label=f'等权组合 (SR={equal_sharpe:.2f})')
for i, symbol in enumerate(SYMBOLS):
ax1.scatter(np.sqrt(cov_matrix.iloc[i, i]) * 100, mean_returns[i] * 100,
marker='x', s=100, label=symbol)
ax1.set_xlabel('年化波动率 (%)')
ax1.set_ylabel('年化收益率 (%)')
ax1.set_title('有效前沿与最优组合', fontsize=12)
ax1.legend(loc='upper left', fontsize=8)
ax1.grid(True, alpha=0.3)
# 图2: 最优权重饼图
ax2 = axes[0, 1]
colors = plt.cm.Set3(np.linspace(0, 1, n_assets))
wedges, texts, autotexts = ax2.pie(optimal_weights, labels=SYMBOLS, autopct='%1.1f%%',
colors=colors, startangle=90)
ax2.set_title('最优组合权重配置', fontsize=12)
# 图3: 相关性热力图
ax3 = axes[1, 0]
im = ax3.imshow(corr_matrix, cmap='RdYlGn', aspect='auto', vmin=-1, vmax=1)
ax3.set_xticks(range(n_assets))
ax3.set_yticks(range(n_assets))
ax3.set_xticklabels(SYMBOLS)
ax3.set_yticklabels(SYMBOLS)
for i in range(n_assets):
for j in range(n_assets):
ax3.text(j, i, f'{corr_matrix.iloc[i, j]:.2f}', ha='center', va='center')
ax3.set_title('资产相关性矩阵', fontsize=12)
plt.colorbar(im, ax=ax3)
# 图4: 个股指标对比
ax4 = axes[1, 1]
x = np.arange(n_assets)
width = 0.35
bars1 = ax4.bar(x - width/2, mean_returns * 100, width, label='年化收益率 (%)')
bars2 = ax4.bar(x + width/2, np.sqrt(np.diag(cov_matrix)) * 100, width, label='年化波动率 (%)')
ax4.set_xticks(x)
ax4.set_xticklabels(SYMBOLS)
ax4.legend()
ax4.set_title('个股收益与风险', fontsize=12)
ax4.grid(True, alpha=0.3, axis='y')
plt.tight_layout()
plt.savefig('charts/portfolio_optimization.png', dpi=150, bbox_inches='tight')
plt.close()
# ========== 6. 输出报告 ==========
print("\n" + "="*60)
print("投资组合优化报告")
print("="*60)
print("\n📊 个股统计")
print("-" * 40)
for i, symbol in enumerate(SYMBOLS):
print(f" {symbol}: 收益 {mean_returns[i]*100:.1f}%, 波动 {np.sqrt(cov_matrix.iloc[i,i])*100:.1f}%")
print("\n🎯 最优组合 (最大夏普比率)")
print("-" * 40)
for i, symbol in enumerate(SYMBOLS):
if optimal_weights[i] > 0.01:
print(f" {symbol}: {optimal_weights[i]*100:.1f}%")
print(f"\n 预期年化收益: {opt_return*100:.1f}%")
print(f" 预期年化波动: {opt_vol*100:.1f}%")
print(f" 夏普比率: {opt_sharpe:.2f}")
print("\n⚖️ 等权组合对比")
print("-" * 40)
print(f" 预期年化收益: {equal_return*100:.1f}%")
print(f" 预期年化波动: {equal_vol*100:.1f}%")
print(f" 夏普比率: {equal_sharpe:.2f}")
print("\n🛡️ 最小方差组合")
print("-" * 40)
for i, symbol in enumerate(SYMBOLS):
if minvar_weights[i] > 0.01:
print(f" {symbol}: {minvar_weights[i]*100:.1f}%")
print(f"\n 预期年化收益: {minvar_return*100:.1f}%")
print(f" 预期年化波动: {minvar_vol*100:.1f}%")
print("\n" + "="*60)
print(f"图表已保存: charts/portfolio_optimization.png")
print("="*60)
```
---
## 量化代码模板库
可直接使用的量化金融代码模板，涵盖数据处理、指标计算、策略开发、回测分析等场景。所有模板遵循最佳实践，包含详细注释。
### 环境配置模板
#### 基础导入
```python
# ========== 量化分析基础导入 ==========
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore')
# 设置显示选项
pd.set_option('display.max_columns', None)
pd.set_option('display.width', None)
pd.set_option('display.float_format', '{:.4f}'.format)
# 设置中文显示
plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans', 'Arial Unicode MS']
plt.rcParams['axes.unicode_minus'] = False
# 设置图表风格
plt.style.use('seaborn-v0_8-whitegrid')
sns.set_palette("husl")
print("环境配置完成！")
```
#### 常用工具函数
```python
# ========== 常用工具函数 ==========
def format_number(num, decimals=2):
"""格式化数字显示"""
if abs(num) >= 1e9:
return f"{num/1e9:.{decimals}f}B"
elif abs(num) >= 1e6:
return f"{num/1e6:.{decimals}f}M"
elif abs(num) >= 1e3:
return f"{num/1e3:.{decimals}f}K"
return f"{num:.{decimals}f}"
def format_percent(num, decimals=2):
"""格式化百分比"""
return f"{num*100:.{decimals}f}%"
def safe_divide(a, b, default=0):
"""安全除法，避免除零错误"""
return a / b if b != 0 else default
def print_dict(d, title=""):
"""美化打印字典"""
if title:
print(f"\n{'='*40}")
print(f" {title}")
print(f"{'='*40}")
for k, v in d.items():
print(f" {k}: {v}")
```
### 数据处理模板
#### 数据清洗
```python
# ========== 数据清洗模板 ==========
def clean_stock_data(df, verbose=True):
"""
股票数据清洗
参数:
df: 原始数据DataFrame
verbose: 是否打印处理信息
返回:
清洗后的DataFrame
"""
df = df.copy()
original_len = len(df)
# 1. 确保日期索引
if not isinstance(df.index, pd.DatetimeIndex):
if 'Date' in df.columns:
df['Date'] = pd.to_datetime(df['Date'])
df.set_index('Date', inplace=True)
elif 'date' in df.columns:
df['date'] = pd.to_datetime(df['date'])
df.set_index('date', inplace=True)
# 2. 排序索引
df.sort_index(inplace=True)
# 3. 删除重复日期
df = df[~df.index.duplicated(keep='last')]
# 4. 处理缺失值
missing_before = df.isnull().sum().sum()
df = df.fillna(method='ffill').fillna(method='bfill')
# 5. 检测并标记异常值（日收益率超过50%）
if 'Close' in df.columns:
df['returns'] = df['Close'].pct_change()
df['is_outlier'] = abs(df['returns']) > 0.5
outlier_count = df['is_outlier'].sum()
df.drop(['returns', 'is_outlier'], axis=1, inplace=True)
else:
outlier_count = 0
# 6. 确保必要的列存在
required_cols = ['Open', 'High', 'Low', 'Close', 'Volume']
for col in required_cols:
if col not in df.columns:
similar = [c for c in df.columns if col.lower() in c.lower()]
if similar:
df[col] = df[similar[0]]
if verbose:
print(f"数据清洗完成:")
print(f" - 原始行数: {original_len}")
print(f" - 清洗后行数: {len(df)}")
print(f" - 处理缺失值: {missing_before}")
print(f" - 异常值数量: {outlier_count}")
print(f" - 日期范围: {df.index[0].date()} ~ {df.index[-1].date()}")
return df
def resample_to_weekly(df):
"""将日线数据转换为周线"""
weekly = df.resample('W').agg({
'Open': 'first',
'High': 'max',
'Low': 'min',
'Close': 'last',
'Volume': 'sum'
}).dropna()
return weekly
def resample_to_monthly(df):
"""将日线数据转换为月线"""
monthly = df.resample('M').agg({
'Open': 'first',
'High': 'max',
'Low': 'min',
'Close': 'last',
'Volume': 'sum'
}).dropna()
return monthly
```
#### 数据统计
```python
# ========== 数据统计模板 ==========
def calculate_basic_stats(df):
"""计算基本统计指标"""
stats = {}
if 'Close' in df.columns:
close = df['Close']
returns = close.pct_change().dropna()
stats['当前价格'] = close.iloc[-1]
stats['期间最高'] = close.max()
stats['期间最低'] = close.min()
stats['平均价格'] = close.mean()
stats['价格标准差'] = close.std()
stats['总收益率'] = format_percent((close.iloc[-1] / close.iloc[0]) - 1)
stats['日均收益率'] = format_percent(returns.mean())
stats['日收益标准差'] = format_percent(returns.std())
stats['年化波动率'] = format_percent(returns.std() * np.sqrt(252))
# 偏度和峰度
stats['偏度'] = f"{returns.skew():.4f}"
stats['峰度'] = f"{returns.kurtosis():.4f}"
if 'Volume' in df.columns:
stats['平均成交量'] = format_number(df['Volume'].mean())
stats['成交量标准差'] = format_number(df['Volume'].std())
return stats
```
### 技术指标模板
#### 趋势指标
```python
# ========== 趋势指标模板 ==========
class TrendIndicators:
"""趋势类技术指标"""
@staticmethod
def SMA(series, window):
"""简单移动平均线"""
return series.rolling(window=window).mean()
@staticmethod
def EMA(series, span):
"""指数移动平均线"""
return series.ewm(span=span, adjust=False).mean()
@staticmethod
def WMA(series, window):
"""加权移动平均线"""
weights = np.arange(1, window + 1)
return series.rolling(window).apply(
lambda x: np.dot(x, weights) / weights.sum(), raw=True
)
@staticmethod
def MACD(series, fast=12, slow=26, signal=9):
"""MACD指标"""
ema_fast = series.ewm(span=fast, adjust=False).mean()
ema_slow = series.ewm(span=slow, adjust=False).mean()
macd_line = ema_fast - ema_slow
signal_line = macd_line.ewm(span=signal, adjust=False).mean()
histogram = macd_line - signal_line
return macd_line, signal_line, histogram
@staticmethod
def ADX(high, low, close, period=14):
"""平均趋向指数"""
tr1 = high - low
tr2 = abs(high - close.shift(1))
tr3 = abs(low - close.shift(1))
tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
atr = tr.rolling(period).mean()
up_move = high - high.shift(1)
down_move = low.shift(1) - low
plus_dm = np.where((up_move > down_move) & (up_move > 0), up_move, 0)
minus_dm = np.where((down_move > up_move) & (down_move > 0), down_move, 0)
plus_di = 100 * pd.Series(plus_dm).rolling(period).mean() / atr
minus_di = 100 * pd.Series(minus_dm).rolling(period).mean() / atr
dx = 100 * abs(plus_di - minus_di) / (plus_di + minus_di)
adx = dx.rolling(period).mean()
return adx, plus_di, minus_di
```
#### 动量指标
```python
# ========== 动量指标模板 ==========
class MomentumIndicators:
"""动量类技术指标"""
@staticmethod
def RSI(series, period=14):
"""相对强弱指标"""
delta = series.diff()
gain = delta.where(delta > 0, 0).rolling(window=period).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
rs = gain / loss
return 100 - (100 / (1 + rs))
@staticmethod
def Stochastic(high, low, close, k_period=14, d_period=3):
"""随机指标KD"""
lowest_low = low.rolling(window=k_period).min()
highest_high = high.rolling(window=k_period).max()
k = 100 * (close - lowest_low) / (highest_high - lowest_low)
d = k.rolling(window=d_period).mean()
return k, d
@staticmethod
def KDJ(high, low, close, n=9, m1=3, m2=3):
"""KDJ指标"""
lowest_low = low.rolling(window=n).min()
highest_high = high.rolling(window=n).max()
rsv = 100 * (close - lowest_low) / (highest_high - lowest_low)
k = rsv.ewm(com=m1-1, adjust=False).mean()
d = k.ewm(com=m2-1, adjust=False).mean()
j = 3 * k - 2 * d
return k, d, j
@staticmethod
def Williams_R(high, low, close, period=14):
"""威廉指标"""
highest_high = high.rolling(window=period).max()
lowest_low = low.rolling(window=period).min()
wr = -100 * (highest_high - close) / (highest_high - lowest_low)
return wr
@staticmethod
def CCI(high, low, close, period=20):
"""顺势指标"""
tp = (high + low + close) / 3
sma_tp = tp.rolling(window=period).mean()
mad = tp.rolling(window=period).apply(lambda x: np.abs(x - x.mean()).mean())
cci = (tp - sma_tp) / (0.015 * mad)
return cci
@staticmethod
def ROC(series, period=12):
"""变动率指标"""
return (series - series.shift(period)) / series.shift(period) * 100
```
#### 波动率指标
```python
# ========== 波动率指标模板 ==========
class VolatilityIndicators:
"""波动率类技术指标"""
@staticmethod
def BollingerBands(series, window=20, num_std=2):
"""布林带"""
middle = series.rolling(window=window).mean()
std = series.rolling(window=window).std()
upper = middle + num_std * std
lower = middle - num_std * std
bandwidth = (upper - lower) / middle
percent_b = (series - lower) / (upper - lower)
return upper, middle, lower, bandwidth, percent_b
@staticmethod
def ATR(high, low, close, period=14):
"""真实波动幅度均值"""
tr1 = high - low
tr2 = abs(high - close.shift(1))
tr3 = abs(low - close.shift(1))
tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
return tr.rolling(window=period).mean()
@staticmethod
def Keltner_Channel(high, low, close, ema_period=20, atr_period=10, multiplier=2):
"""肯特纳通道"""
typical_price = (high + low + close) / 3
middle = typical_price.ewm(span=ema_period, adjust=False).mean()
atr = VolatilityIndicators.ATR(high, low, close, atr_period)
upper = middle + multiplier * atr
lower = middle - multiplier * atr
return upper, middle, lower
@staticmethod
def Donchian_Channel(high, low, period=20):
"""唐奇安通道"""
upper = high.rolling(window=period).max()
lower = low.rolling(window=period).min()
middle = (upper + lower) / 2
return upper, middle, lower
@staticmethod
def Historical_Volatility(series, window=20):
"""历史波动率（年化）"""
log_returns = np.log(series / series.shift(1))
return log_returns.rolling(window=window).std() * np.sqrt(252)
```
#### 成交量指标
```python
# ========== 成交量指标模板 ==========
class VolumeIndicators:
"""成交量类技术指标"""
@staticmethod
def OBV(close, volume):
"""能量潮指标"""
obv = pd.Series(index=close.index, dtype=float)
obv.iloc[0] = volume.iloc[0]
for i in range(1, len(close)):
if close.iloc[i] > close.iloc[i-1]:
obv.iloc[i] = obv.iloc[i-1] + volume.iloc[i]
elif close.iloc[i] < close.iloc[i-1]:
obv.iloc[i] = obv.iloc[i-1] - volume.iloc[i]
else:
obv.iloc[i] = obv.iloc[i-1]
return obv
@staticmethod
def VWAP(high, low, close, volume):
"""成交量加权平均价"""
typical_price = (high + low + close) / 3
return (typical_price * volume).cumsum() / volume.cumsum()
@staticmethod
def Volume_MA(volume, window=20):
"""成交量移动平均"""
return volume.rolling(window=window).mean()
@staticmethod
def Volume_Ratio(volume, window=5):
"""量比"""
avg_volume = volume.shift(1).rolling(window=window).mean()
return volume / avg_volume
@staticmethod
def MFI(high, low, close, volume, period=14):
"""资金流量指标"""
typical_price = (high + low + close) / 3
raw_money_flow = typical_price * volume
positive_flow = pd.Series(0, index=close.index, dtype=float)
negative_flow = pd.Series(0, index=close.index, dtype=float)
for i in range(1, len(typical_price)):
if typical_price.iloc[i] > typical_price.iloc[i-1]:
positive_flow.iloc[i] = raw_money_flow.iloc[i]
else:
negative_flow.iloc[i] = raw_money_flow.iloc[i]
positive_mf = positive_flow.rolling(window=period).sum()
negative_mf = negative_flow.rolling(window=period).sum()
mfi = 100 - (100 / (1 + positive_mf / negative_mf))
return mfi
```
### 策略框架模板
#### 策略基类
```python
# ========== 策略基类模板 ==========
from abc import ABC, abstractmethod
class BaseStrategy(ABC):
"""策略基类"""
def __init__(self, name="BaseStrategy"):
self.name = name
self.signals = None
self.positions = None
@abstractmethod
def generate_signals(self, df):
"""生成交易信号（子类必须实现）"""
pass
def backtest(self, df, initial_capital=100000, commission=0.001):
"""回测策略"""
signals = self.generate_signals(df)
# 计算持仓
signals['position'] = signals['signal'].shift(1).fillna(0)
# 计算收益
signals['returns'] = df['Close'].pct_change()
signals['strategy_returns'] = signals['position'] * signals['returns']
# 扣除交易成本
signals['trades'] = signals['position'].diff().abs()
signals['strategy_returns'] -= signals['trades'] * commission
# 计算累计收益
signals['cumulative_returns'] = (1 + signals['returns']).cumprod()
signals['cumulative_strategy'] = (1 + signals['strategy_returns']).cumprod()
# 计算组合价值
signals['portfolio_value'] = initial_capital * signals['cumulative_strategy']
self.signals = signals
return signals
def get_performance_summary(self):
"""获取绩效摘要"""
if self.signals is None:
raise ValueError("请先运行回测")
returns = self.signals['strategy_returns'].dropna()
total_return = (1 + returns).prod() - 1
annual_return = (1 + total_return) ** (252 / len(returns)) - 1
volatility = returns.std() * np.sqrt(252)
sharpe = annual_return / volatility if volatility != 0 else 0
cumulative = (1 + returns).cumprod()
peak = cumulative.expanding().max()
drawdown = (cumulative - peak) / peak
max_dd = drawdown.min()
win_trades = (returns > 0).sum()
total_trades = (returns != 0).sum()
win_rate = win_trades / total_trades if total_trades > 0 else 0
return {
'策略名称': self.name,
'总收益率': f"{total_return:.2%}",
'年化收益率': f"{annual_return:.2%}",
'年化波动率': f"{volatility:.2%}",
'夏普比率': f"{sharpe:.2f}",
'最大回撤': f"{max_dd:.2%}",
'胜率': f"{win_rate:.2%}",
'交易次数': total_trades
}
```
#### 常用策略实现
```python
# ========== 常用策略模板 ==========
class DualMovingAverageStrategy(BaseStrategy):
"""双均线策略"""
def __init__(self, short_window=10, long_window=30):
super().__init__(name=f"DualMA({short_window},{long_window})")
self.short_window = short_window
self.long_window = long_window
def generate_signals(self, df):
signals = pd.DataFrame(index=df.index)
signals['price'] = df['Close']
signals['short_ma'] = df['Close'].rolling(self.short_window).mean()
signals['long_ma'] = df['Close'].rolling(self.long_window).mean()
signals['signal'] = 0
signals.loc[signals['short_ma'] > signals['long_ma'], 'signal'] = 1
signals.loc[signals['short_ma'] < signals['long_ma'], 'signal'] = -1
return signals
class RSIMeanReversionStrategy(BaseStrategy):
"""RSI均值回归策略"""
def __init__(self, period=14, oversold=30, overbought=70):
super().__init__(name=f"RSI({period},{oversold},{overbought})")
self.period = period
self.oversold = oversold
self.overbought = overbought
def generate_signals(self, df):
signals = pd.DataFrame(index=df.index)
signals['price'] = df['Close']
signals['RSI'] = MomentumIndicators.RSI(df['Close'], self.period)
signals['signal'] = 0
signals.loc[signals['RSI'] < self.oversold, 'signal'] = 1
signals.loc[signals['RSI'] > self.overbought, 'signal'] = -1
# 保持仓位直到反向信号
signals['signal'] = signals['signal'].replace(0, np.nan).ffill().fillna(0)
return signals
class BollingerBandsStrategy(BaseStrategy):
"""布林带突破策略"""
def __init__(self, window=20, num_std=2):
super().__init__(name=f"BB({window},{num_std})")
self.window = window
self.num_std = num_std
def generate_signals(self, df):
signals = pd.DataFrame(index=df.index)
signals['price'] = df['Close']
upper, middle, lower, _, _ = VolatilityIndicators.BollingerBands(
df['Close'], self.window, self.num_std
)
signals['upper'] = upper
signals['middle'] = middle
signals['lower'] = lower
signals['signal'] = 0
signals.loc[signals['price'] < signals['lower'], 'signal'] = 1 # 超卖买入
signals.loc[signals['price'] > signals['upper'], 'signal'] = -1 # 超买卖出
signals['signal'] = signals['signal'].replace(0, np.nan).ffill().fillna(0)
return signals
class MomentumStrategy(BaseStrategy):
"""动量策略"""
def __init__(self, lookback=20, threshold=0):
super().__init__(name=f"Momentum({lookback})")
self.lookback = lookback
self.threshold = threshold
def generate_signals(self, df):
signals = pd.DataFrame(index=df.index)
signals['price'] = df['Close']
signals['momentum'] = df['Close'].pct_change(self.lookback)
signals['signal'] = 0
signals.loc[signals['momentum'] > self.threshold, 'signal'] = 1
signals.loc[signals['momentum'] < -self.threshold, 'signal'] = -1
return signals
```
### 绩效分析模板
```python
# ========== 绩效分析模板 ==========
class PerformanceMetrics:
"""绩效指标计算"""
def __init__(self, returns, benchmark_returns=None, risk_free_rate=0.02):
self.returns = returns.dropna()
self.benchmark = benchmark_returns
self.rf = risk_free_rate / 252
def all_metrics(self):
"""计算所有指标"""
return {
'总收益率': self.total_return(),
'年化收益率': self.annualized_return(),
'年化波动率': self.annualized_volatility(),
'夏普比率': self.sharpe_ratio(),
'索提诺比率': self.sortino_ratio(),
'卡尔玛比率': self.calmar_ratio(),
'最大回撤': self.max_drawdown(),
'最大回撤持续期': self.max_drawdown_duration(),
'胜率': self.win_rate(),
'盈亏比': self.profit_loss_ratio(),
'日均收益': self.returns.mean(),
'日收益标准差': self.returns.std(),
'偏度': self.returns.skew(),
'峰度': self.returns.kurtosis(),
'正收益天数': (self.returns > 0).sum(),
'负收益天数': (self.returns < 0).sum(),
}
def total_return(self):
return (1 + self.returns).prod() - 1
def annualized_return(self):
total_days = len(self.returns)
if total_days == 0:
return 0
return (1 + self.total_return()) ** (252 / total_days) - 1
def annualized_volatility(self):
return self.returns.std() * np.sqrt(252)
def sharpe_ratio(self):
excess = self.returns - self.rf
if excess.std() == 0:
return 0
return np.sqrt(252) * excess.mean() / excess.std()
def sortino_ratio(self):
excess = self.returns - self.rf
downside = self.returns[self.returns < 0]
if len(downside) == 0 or downside.std() == 0:
return np.inf
downside_std = downside.std() * np.sqrt(252)
return self.annualized_return() / downside_std
def calmar_ratio(self):
max_dd = abs(self.max_drawdown())
if max_dd == 0:
return np.inf
return self.annualized_return() / max_dd
def max_drawdown(self):
cumulative = (1 + self.returns).cumprod()
peak = cumulative.expanding().max()
drawdown = (cumulative - peak) / peak
return drawdown.min()
def max_drawdown_duration(self):
cumulative = (1 + self.returns).cumprod()
peak = cumulative.expanding().max()
drawdown = (cumulative - peak) / peak
in_drawdown = drawdown < 0
duration = 0
max_duration = 0
for is_dd in in_drawdown:
if is_dd:
duration += 1
max_duration = max(max_duration, duration)
else:
duration = 0
return max_duration
def win_rate(self):
wins = (self.returns > 0).sum()
total = (self.returns != 0).sum()
return wins / total if total > 0 else 0
def profit_loss_ratio(self):
avg_win = self.returns[self.returns > 0].mean()
avg_loss = abs(self.returns[self.returns < 0].mean())
if avg_loss == 0:
return np.inf
return avg_win / avg_loss
```
### 回测报告可视化模板
```python
# ========== 回测报告模板 ==========
def generate_backtest_report(strategy, df, save_path=None):
"""生成完整回测报告"""
# 运行回测
signals = strategy.backtest(df)
# 计算绩效指标
metrics = PerformanceMetrics(signals['strategy_returns'].dropna())
all_metrics = metrics.all_metrics()
# 创建图表
fig = plt.figure(figsize=(16, 12))
# 子图1：累计收益对比
ax1 = fig.add_subplot(3, 2, 1)
ax1.plot(signals['cumulative_returns'], label='买入持有', alpha=0.7)
ax1.plot(signals['cumulative_strategy'], label='策略收益', alpha=0.7)
ax1.set_title('累计收益对比')
ax1.legend()
ax1.grid(True, alpha=0.3)
# 子图2：回撤分析
ax2 = fig.add_subplot(3, 2, 2)
cumulative = signals['cumulative_strategy']
peak = cumulative.expanding().max()
drawdown = (cumulative - peak) / peak
ax2.fill_between(drawdown.index, drawdown.values, 0, alpha=0.5, color='red')
ax2.set_title('回撤分析')
ax2.grid(True, alpha=0.3)
# 子图3：日收益分布
ax3 = fig.add_subplot(3, 2, 3)
signals['strategy_returns'].hist(bins=50, ax=ax3, alpha=0.7)
ax3.axvline(x=0, color='red', linestyle='--')
ax3.set_title('日收益分布')
ax3.grid(True, alpha=0.3)
# 子图4：月度收益热力图
ax4 = fig.add_subplot(3, 2, 4)
monthly_returns = signals['strategy_returns'].resample('M').sum()
monthly_df = pd.DataFrame({
'year': monthly_returns.index.year,
'month': monthly_returns.index.month,
'return': monthly_returns.values
})
pivot = monthly_df.pivot(index='year', columns='month', values='return')
sns.heatmap(pivot, annot=True, fmt='.1%', cmap='RdYlGn', center=0, ax=ax4)
ax4.set_title('月度收益热力图')
# 子图5：交易信号
ax5 = fig.add_subplot(3, 2, 5)
ax5.plot(signals['price'], label='价格', alpha=0.7)
buy_signals = signals[signals['signal'] == 1]
sell_signals = signals[signals['signal'] == -1]
ax5.scatter(buy_signals.index, buy_signals['price'], marker='^',
color='green', label='买入', alpha=0.7, s=50)
ax5.scatter(sell_signals.index, sell_signals['price'], marker='v',
color='red', label='卖出', alpha=0.7, s=50)
ax5.set_title('交易信号')
ax5.legend()
ax5.grid(True, alpha=0.3)
# 子图6：绩效指标表格
ax6 = fig.add_subplot(3, 2, 6)
ax6.axis('off')
metrics_text = f"""
═══════════════════════════════════
策略: {strategy.name}
═══════════════════════════════════
总收益率: {all_metrics['总收益率']:.2%}
年化收益率: {all_metrics['年化收益率']:.2%}
年化波动率: {all_metrics['年化波动率']:.2%}
夏普比率: {all_metrics['夏普比率']:.2f}
索提诺比率: {all_metrics['索提诺比率']:.2f}
最大回撤: {all_metrics['最大回撤']:.2%}
胜率: {all_metrics['胜率']:.2%}
盈亏比: {all_metrics['盈亏比']:.2f}
═══════════════════════════════════
"""
ax6.text(0.1, 0.5, metrics_text, transform=ax6.transAxes,
fontsize=10, verticalalignment='center', fontfamily='monospace')
plt.tight_layout()
if save_path:
plt.savefig(save_path, dpi=150, bbox_inches='tight')
print(f"报告已保存至: {save_path}")
plt.show()
return all_metrics
```
### K线图模板
```python
# ========== K线图模板 ==========
def plot_candlestick_chart(df, title="K线图", indicators=None, save_path=None):
"""
绑制带指标的K线图
参数:
df: 包含OHLCV数据的DataFrame
title: 图表标题
indicators: 要添加的指标列表 ['MA', 'BOLL', 'MACD', 'RSI', 'VOL']
save_path: 保存路径
"""
if indicators is None:
indicators = ['MA', 'VOL']
# 计算需要的子图数量
n_subplots = 1 # 主图
if 'VOL' in indicators:
n_subplots += 1
if 'MACD' in indicators:
n_subplots += 1
if 'RSI' in indicators:
n_subplots += 1
# 设置子图高度比例
heights = [3]
if 'VOL' in indicators:
heights.append(1)
if 'MACD' in indicators:
heights.append(1)
if 'RSI' in indicators:
heights.append(1)
fig, axes = plt.subplots(n_subplots, 1, figsize=(14, 3*n_subplots),
gridspec_kw={'height_ratios': heights})
if n_subplots == 1:
axes = [axes]
ax_main = axes[0]
current_ax = 1
# 主图：价格走势
ax_main.plot(df.index, df['Close'], label='收盘价', color='black', linewidth=1)
ax_main.fill_between(df.index, df['Low'], df['High'], alpha=0.2, color='gray')
# 添加均线
if 'MA' in indicators:
for window, color in [(5, 'red'), (20, 'blue'), (60, 'green')]:
ma = df['Close'].rolling(window).mean()
ax_main.plot(df.index, ma, label=f'MA{window}', color=color,
linestyle='--', linewidth=0.8)
# 添加布林带
if 'BOLL' in indicators:
upper, middle, lower, _, _ = VolatilityIndicators.BollingerBands(df['Close'])
ax_main.plot(df.index, upper, 'b--', linewidth=0.5, alpha=0.5)
ax_main.plot(df.index, lower, 'b--', linewidth=0.5, alpha=0.5)
ax_main.fill_between(df.index, upper, lower, alpha=0.1, color='blue')
ax_main.set_title(title, fontsize=14)
ax_main.set_ylabel('价格')
ax_main.legend(loc='upper left')
ax_main.grid(True, alpha=0.3)
# 成交量图
if 'VOL' in indicators:
ax_vol = axes[current_ax]
colors = ['red' if df['Close'].iloc[i] >= df['Open'].iloc[i]
else 'green' for i in range(len(df))]
ax_vol.bar(df.index, df['Volume'], color=colors, alpha=0.5)
ax_vol.set_ylabel('成交量')
ax_vol.grid(True, alpha=0.3)
current_ax += 1
# MACD图
if 'MACD' in indicators:
ax_macd = axes[current_ax]
macd, signal, hist = TrendIndicators.MACD(df['Close'])
ax_macd.plot(df.index, macd, label='MACD', color='blue', linewidth=0.8)
ax_macd.plot(df.index, signal, label='Signal', color='orange', linewidth=0.8)
colors = ['red' if h >= 0 else 'green' for h in hist]
ax_macd.bar(df.index, hist, color=colors, alpha=0.5)
ax_macd.axhline(y=0, color='gray', linestyle='-', linewidth=0.5)
ax_macd.set_ylabel('MACD')
ax_macd.legend(loc='upper left')
ax_macd.grid(True, alpha=0.3)
current_ax += 1
# RSI图
if 'RSI' in indicators:
ax_rsi = axes[current_ax]
rsi = MomentumIndicators.RSI(df['Close'])
ax_rsi.plot(df.index, rsi, color='purple', linewidth=0.8)
ax_rsi.axhline(y=70, color='red', linestyle='--', linewidth=0.5)
ax_rsi.axhline(y=30, color='green', linestyle='--', linewidth=0.5)
ax_rsi.fill_between(df.index, 30, 70, alpha=0.1, color='gray')
ax_rsi.set_ylabel('RSI')
ax_rsi.set_ylim(0, 100)
ax_rsi.grid(True, alpha=0.3)
plt.xlabel('日期')
plt.tight_layout()
if save_path:
plt.savefig(save_path, dpi=150, bbox_inches='tight')
print(f"图表已保存至: {save_path}")
plt.show()
```
---
## 技术分析指南
技术分析是通过研究历史价格和交易量数据来预测未来价格走势的方法。
### 趋势分析
#### 移动平均线系统
**简单移动平均线 (SMA)**
- 短期：5日、10日
- 中期：20日、50日
- 长期：100日、200日
**指数移动平均线 (EMA)**
- 对近期价格赋予更高权重
- 比SMA更快反应价格变化
**金叉与死叉**
- 金叉：短期均线向上穿越长期均线（买入信号）
- 死叉：短期均线向下穿越长期均线（卖出信号）
#### 趋势线和通道
- 上升趋势线：连接更高的低点
- 下降趋势线：连接更低的高点
- 通道：平行的趋势线形成的价格区间
### 动量指标解读
#### RSI (相对强弱指标)
```python
def interpret_rsi(rsi_value):
if rsi_value > 70:
return "超买区域，可能回调"
elif rsi_value < 30:
return "超卖区域，可能反弹"
else:
return "中性区域"
```
**使用规则**：
- RSI > 70：超买，考虑卖出
- RSI < 30：超卖，考虑买入
- RSI背离：价格与RSI方向相反，可能反转
#### MACD (移动平均收敛发散)
**组成部分**：
- MACD线：12日EMA - 26日EMA
- 信号线：MACD的9日EMA
- 柱状图：MACD线 - 信号线
**交易信号**：
- MACD线上穿信号线：买入信号
- MACD线下穿信号线：卖出信号
- 零轴上方：多头市场
- 零轴下方：空头市场
#### 随机指标 (KDJ)
- K线：快速随机指标
- D线：K线的3日平均
- J线：3K - 2D
**使用规则**：
- K > 80, D > 80：超买
- K < 20, D < 20：超卖
- K上穿D：买入信号
- K下穿D：卖出信号
### 波动率指标解读
#### 布林带 (Bollinger Bands)
**计算方法**：
- 中轨：20日SMA
- 上轨：中轨 + 2倍标准差
- 下轨：中轨 - 2倍标准差
**交易策略**：
- 价格触及上轨：可能超买
- 价格触及下轨：可能超卖
- 带宽收窄：预示大幅波动
- 带宽扩张：趋势正在形成
#### ATR (真实波动幅度)
**用途**：
- 设置止损位
- 确定头寸规模
- 衡量市场波动性
### 成交量分析
#### 量价关系
| 价格 | 成交量 | 含义 |
|------|--------|------|
| 上涨 | 放量 | 上涨动能强 |
| 上涨 | 缩量 | 上涨乏力 |
| 下跌 | 放量 | 下跌动能强 |
| 下跌 | 缩量 | 下跌乏力 |
#### OBV (能量潮指标)
- 价格上涨日：OBV += 当日成交量
- 价格下跌日：OBV -= 当日成交量
- OBV趋势与价格趋势背离时注意反转
### K线形态
#### 反转形态
**看涨形态**：
- 锤子线：下影线长，上影线短或无
- 看涨吞没：大阳线吞没前一根阴线
- 早晨之星：三根K线组合
**看跌形态**：
- 吊颈线：与锤子线相同但出现在上涨顶部
- 看跌吞没：大阴线吞没前一根阳线
- 黄昏之星：三根K线组合
#### 持续形态
- 上升三法：上涨中继形态
- 下降三法：下跌中继形态
### 支撑与阻力
#### 识别方法
1. **历史高低点**：前期高点形成阻力，前期低点形成支撑
2. **整数关卡**：心理价位如100、500、1000等
3. **移动平均线**：重要均线如200日均线
4. **斐波那契回撤**：23.6%、38.2%、50%、61.8%
#### 斐波那契应用
```python
def fibonacci_levels(high, low):
"""计算斐波那契回撤位"""
diff = high - low
levels = {
'0%': low,
'23.6%': low + 0.236 * diff,
'38.2%': low + 0.382 * diff,
'50%': low + 0.5 * diff,
'61.8%': low + 0.618 * diff,
'100%': high
}
return levels
```
### 综合技术分析框架
进行技术分析时，建议按以下步骤：
1. **确定大趋势**：查看周线和日线图
2. **识别支撑阻力**：标记关键价位
3. **分析动量指标**：RSI、MACD等
4. **观察量价关系**：确认趋势强度
5. **寻找入场时机**：结合多个信号
6. **设定止损止盈**：使用ATR或支撑阻力位
---
## 基本面分析
基本面分析通过研究公司的财务状况、行业地位和宏观经济环境来评估其内在价值。
### 财务报表分析
#### 1. 资产负债表分析
**流动性指标**
```python
def liquidity_ratios(current_assets, current_liabilities, inventory, cash):
"""计算流动性指标"""
return {
'流动比率': current_assets / current_liabilities,
'速动比率': (current_assets - inventory) / current_liabilities,
'现金比率': cash / current_liabilities
}
```
**健康标准**：
- 流动比率 > 2：良好
- 速动比率 > 1：良好
- 现金比率 > 0.5：良好
**资本结构**
```python
def leverage_ratios(total_debt, total_equity, total_assets, ebit, interest):
"""计算杠杆指标"""
return {
'资产负债率': total_debt / total_assets,
'权益乘数': total_assets / total_equity,
'利息保障倍数': ebit / interest
}
```
#### 2. 利润表分析
**盈利能力指标**
```python
def profitability_ratios(gross_profit, operating_income, net_income, revenue, total_assets, equity):
"""计算盈利能力指标"""
return {
'毛利率': gross_profit / revenue,
'营业利润率': operating_income / revenue,
'净利率': net_income / revenue,
'ROA': net_income / total_assets,
'ROE': net_income / equity
}
```
**ROE杜邦分析**
```
ROE = 净利率 × 资产周转率 × 权益乘数
= (净利润/营收) × (营收/资产) × (资产/股东权益)
```
#### 3. 现金流量表分析
**关键指标**
- 经营活动现金流：反映主营业务造血能力
- 自由现金流 = 经营现金流 - 资本支出
- 现金流量比率 = 经营现金流 / 流动负债
**现金流质量评估**
```python
def cash_flow_quality(net_income, operating_cash_flow, capex):
"""评估现金流质量"""
fcf = operating_cash_flow - capex
ocf_to_ni = operating_cash_flow / net_income
quality = "优秀" if ocf_to_ni > 1.2 else "良好" if ocf_to_ni > 0.8 else "一般"
return {
'自由现金流': fcf,
'经营现金流/净利润': ocf_to_ni,
'现金流质量': quality
}
```
### 估值方法
#### 1. 相对估值法
**市盈率 (P/E)**
```python
def pe_valuation(earnings_per_share, industry_pe, growth_rate):
"""P/E估值"""
fair_value_static = earnings_per_share * industry_pe
peg_fair_pe = growth_rate * 100
fair_value_peg = earnings_per_share * min(peg_fair_pe, industry_pe * 1.5)
return {
'静态估值': fair_value_static,
'PEG估值': fair_value_peg
}
```
**估值指标对比**
| 指标 | 适用场景 | 局限性 |
|------|----------|--------|
| P/E | 盈利稳定的公司 | 不适用于亏损公司 |
| P/B | 资产密集型行业 | 不适用于轻资产公司 |
| P/S | 高增长但未盈利公司 | 忽略盈利能力 |
| EV/EBITDA | 跨国比较 | 忽略资本支出 |
#### 2. 绝对估值法
**DCF模型**
```python
def dcf_valuation(free_cash_flows, wacc, terminal_growth, shares_outstanding):
"""DCF估值模型"""
pv_fcf = sum([fcf / (1 + wacc) ** (i + 1)
for i, fcf in enumerate(free_cash_flows)])
terminal_value = free_cash_flows[-1] * (1 + terminal_growth) / (wacc - terminal_growth)
pv_terminal = terminal_value / (1 + wacc) ** len(free_cash_flows)
enterprise_value = pv_fcf + pv_terminal
equity_value_per_share = enterprise_value / shares_outstanding
return {
'预测期现值': pv_fcf,
'终值现值': pv_terminal,
'企业价值': enterprise_value,
'每股价值': equity_value_per_share
}
```
**WACC计算**
```python
def calculate_wacc(equity_value, debt_value, cost_of_equity, cost_of_debt, tax_rate):
"""计算加权平均资本成本"""
total_value = equity_value + debt_value
wacc = (equity_value / total_value * cost_of_equity +
debt_value / total_value * cost_of_debt * (1 - tax_rate))
return wacc
```
#### 3. 股息折现模型
**Gordon增长模型**
```python
def gordon_growth_model(dividend, growth_rate, required_return):
"""Gordon增长模型"""
if required_return <= growth_rate:
raise ValueError("要求回报率必须大于增长率")
fair_value = dividend * (1 + growth_rate) / (required_return - growth_rate)
return fair_value
```
### 行业分析
#### 波特五力模型
1. **现有竞争者的威胁** — 行业集中度、竞争强度
2. **新进入者的威胁** — 进入壁垒高度、规模经济
3. **替代品的威胁** — 替代品可用性、转换成本
4. **供应商的议价能力** — 供应商集中度、关键原材料依赖
5. **买方的议价能力** — 客户集中度、产品差异化程度
#### 行业生命周期
| 阶段 | 特征 | 投资策略 |
|------|------|----------|
| 初创期 | 高增长、高风险 | 关注技术和市场潜力 |
| 成长期 | 快速增长、竞争加剧 | 关注市场份额和增长率 |
| 成熟期 | 稳定增长、高分红 | 关注现金流和股息 |
| 衰退期 | 负增长、行业整合 | 谨慎投资，关注转型 |
### 综合基本面分析框架
1. **宏观环境分析** — 经济周期位置、利率环境、政策导向
2. **行业分析** — 行业增长前景、竞争格局、监管环境
3. **公司分析** — 商业模式、竞争优势（护城河）、管理团队
4. **财务分析** — 盈利能力、财务健康度、现金流状况
5. **估值分析** — 相对估值、绝对估值、安全边际
### 投资检查清单
- [ ] 是否理解公司的商业模式？
- [ ] 公司是否有持续的竞争优势？
- [ ] 财务状况是否健康？
- [ ] 现金流是否充足？
- [ ] 估值是否合理？
- [ ] 管理层是否值得信赖？
- [ ] 行业前景如何？
- [ ] 主要风险是什么？
---
## 风险管理
### 风险识别
- **市场风险**：资产价格波动导致的损失
- **信用风险**：交易对手违约风险
- **流动性风险**：无法及时变现的风险
- **操作风险**：内部流程或系统故障
### 风险指标计算
#### VaR计算
```python
def calculate_var(returns, confidence_level=0.95, method='historical'):
"""计算风险价值(VaR)"""
if method == 'historical':
var = np.percentile(returns, (1 - confidence_level) * 100)
elif method == 'parametric':
from scipy import stats
mean = returns.mean()
std = returns.std()
z_score = stats.norm.ppf(1 - confidence_level)
var = mean + z_score * std
return abs(var)
def calculate_cvar(returns, confidence_level=0.95):
"""计算条件风险价值(CVaR/ES)"""
var = calculate_var(returns, confidence_level)
cvar = returns[returns <= -var].mean()
return abs(cvar)
```
#### 最大回撤
```python
def calculate_max_drawdown(cumulative_returns):
"""计算最大回撤"""
peak = cumulative_returns.expanding(min_periods=1).max()
drawdown = (cumulative_returns - peak) / peak
max_dd = drawdown.min()
return abs(max_dd)
```
#### 风险调整收益
```python
def calculate_sharpe_ratio(returns, risk_free_rate=0.02):
"""计算夏普比率"""
excess_returns = returns - risk_free_rate / 252
return np.sqrt(252) * excess_returns.mean() / excess_returns.std()
def calculate_sortino_ratio(returns, risk_free_rate=0.02):
"""计算索提诺比率"""
excess_returns = returns - risk_free_rate / 252
downside_returns = returns[returns < 0]
downside_std = downside_returns.std()
return np.sqrt(252) * excess_returns.mean() / downside_std
```
#### 波动率与Beta
```python
def calculate_volatility(returns, window=20):
"""计算滚动波动率"""
return returns.rolling(window=window).std() * np.sqrt(252)
def calculate_beta(stock_returns, market_returns):
"""计算Beta系数"""
covariance = np.cov(stock_returns, market_returns)[0][1]
market_variance = np.var(market_returns)
return covariance / market_variance
```
### 风险阈值标准
| 指标 | 低风险 | 中风险 | 高风险 |
|------|--------|--------|--------|
| 日VaR (95%) | <1% | 1-3% | >3% |
| 最大回撤 | <10% | 10-25% | >25% |
| 年化波动率 | <15% | 15-30% | >30% |
| Beta | <0.8 | 0.8-1.2 | >1.2 |
### 风险报告格式
标准风险报告应包含：
1. **风险摘要** — 当前风险水平评估（低/中/高）、主要风险因素列表
2. **定量指标** — VaR (95%)、最大回撤、波动率、夏普比率、Beta
3. **压力测试结果** — 历史情景分析、假设情景分析
4. **风险管理建议** — 具体行动建议、对冲策略推荐
---
## 投资组合优化
### 现代投资组合理论
#### 马科维茨均值-方差优化
**核心概念**
- 预期收益：资产的平均回报
- 风险：收益的标准差
- 相关性：资产间的联动关系
- 有效前沿：给定风险下收益最高的组合集合
```python
def portfolio_optimization(returns, cov_matrix, risk_free_rate=0.02):
"""马科维茨投资组合优化"""
n_assets = len(returns)
def portfolio_return(weights):
return np.dot(weights, returns)
def portfolio_volatility(weights):
return np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights)))
def sharpe_ratio(weights):
ret = portfolio_return(weights)
vol = portfolio_volatility(weights)
return -(ret - risk_free_rate) / vol
constraints = ({'type': 'eq', 'fun': lambda x: np.sum(x) - 1})
bounds = tuple((0, 1) for _ in range(n_assets))
init_weights = np.array([1/n_assets] * n_assets)
result = minimize(sharpe_ratio, init_weights, method='SLSQP',
bounds=bounds, constraints=constraints)
return result.x
```
**有效前沿计算**
```python
def efficient_frontier(returns, cov_matrix, n_points=100):
"""计算有效前沿"""
n_assets = len(returns)
results = {'return': [], 'volatility': [], 'weights': []}
target_returns = np.linspace(returns.min(), returns.max(), n_points)
for target in target_returns:
constraints = [
{'type': 'eq', 'fun': lambda x: np.sum(x) - 1},
{'type': 'eq', 'fun': lambda x, t=target: np.dot(x, returns) - t}
]
bounds = tuple((0, 1) for _ in range(n_assets))
init = np.array([1/n_assets] * n_assets)
result = minimize(
lambda w: np.sqrt(np.dot(w.T, np.dot(cov_matrix, w))),
init, method='SLSQP', bounds=bounds, constraints=constraints
)
if result.success:
results['return'].append(target)
results['volatility'].append(result.fun)
results['weights'].append(result.x)
return results
```
#### Black-Litterman模型
将市场均衡收益与投资者观点相结合，生成更稳健的预期收益估计。
```python
def black_litterman(market_weights, sigma, tau, P, Q, omega, risk_aversion=2.5):
"""Black-Litterman模型"""
pi = risk_aversion * np.dot(sigma, market_weights)
tau_sigma = tau * sigma
M_inverse = np.linalg.inv(np.linalg.inv(tau_sigma) + np.dot(np.dot(P.T, np.linalg.inv(omega)), P))
posterior_return = np.dot(M_inverse,
np.dot(np.linalg.inv(tau_sigma), pi) +
np.dot(np.dot(P.T, np.linalg.inv(omega)), Q))
return posterior_return
```
#### 风险平价策略
每个资产对组合总风险的贡献相等。
```python
def risk_parity_weights(cov_matrix):
"""风险平价权重计算"""
n_assets = len(cov_matrix)
def risk_contribution(weights):
portfolio_vol = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights)))
marginal_contrib = np.dot(cov_matrix, weights)
risk_contrib = weights * marginal_contrib / portfolio_vol
return risk_contrib
def objective(weights):
rc = risk_contribution(weights)
target_rc = 1 / n_assets
return np.sum((rc - target_rc) ** 2)
constraints = {'type': 'eq', 'fun': lambda x: np.sum(x) - 1}
bounds = tuple((0.01, 1) for _ in range(n_assets))
init = np.array([1/n_assets] * n_assets)
result = minimize(objective, init, method='SLSQP',
bounds=bounds, constraints=constraints)
return result.x
```
### 资产配置策略
#### 战略资产配置
**典型配置模板**
| 风险等级 | 股票 | 债券 | 另类投资 | 现金 |
|----------|------|------|----------|------|
| 保守型 | 20% | 60% | 10% | 10% |
| 稳健型 | 40% | 45% | 10% | 5% |
| 平衡型 | 60% | 30% | 8% | 2% |
| 进取型 | 80% | 15% | 5% | 0% |
| 激进型 | 95% | 0% | 5% | 0% |
#### 战术资产配置
| 市场环境 | 调整建议 |
|----------|----------|
| 经济扩张 | 超配股票，低配债券 |
| 经济衰退 | 低配股票，超配债券 |
| 高通胀 | 超配大宗商品、TIPS |
| 低通胀 | 标准配置 |
| 高波动 | 降低股票仓位，增加现金 |
#### 核心-卫星策略
```
总组合 = 核心组合 (70-80%) + 卫星组合 (20-30%)
```
- **核心组合**：低成本指数基金，提供市场Beta
- **卫星组合**：主动管理，追求Alpha
### 组合再平衡
```python
def check_rebalance(current_weights, target_weights, threshold=0.05):
"""检查是否需要再平衡"""
deviation = np.abs(current_weights - target_weights)
max_deviation = np.max(deviation)
needs_rebalance = max_deviation > threshold
return {
'needs_rebalance': needs_rebalance,
'max_deviation': max_deviation,
'deviations': dict(zip(range(len(deviation)), deviation))
}
```
#### 再平衡策略
| 策略 | 触发条件 | 优点 | 缺点 |
|------|----------|------|------|
| 日历再平衡 | 固定时间间隔 | 简单规则化 | 可能错过最佳时机 |
| 阈值再平衡 | 偏离度超过阈值 | 响应市场变化 | 可能频繁交易 |
| 混合再平衡 | 日历+阈值 | 平衡两者优点 | 规则更复杂 |
#### 税务优化再平衡
```python
def tax_efficient_rebalance(current_weights, target_weights,
unrealized_gains, tax_rate=0.2):
"""税务优化的再平衡"""
adjustments = target_weights - current_weights
sell_priority = np.argsort(unrealized_gains)
tax_cost = np.maximum(unrealized_gains, 0) * tax_rate
return {
'adjustments': adjustments,
'sell_priority': sell_priority,
'estimated_tax': np.sum(tax_cost[adjustments < 0])
}
```
### 绩效归因
#### Brinson归因模型
```python
def brinson_attribution(portfolio_weights, benchmark_weights,
portfolio_returns, benchmark_returns):
"""Brinson归因分析"""
allocation = np.sum((portfolio_weights - benchmark_weights) * benchmark_returns)
selection = np.sum(benchmark_weights * (portfolio_returns - benchmark_returns))
interaction = np.sum((portfolio_weights - benchmark_weights) *
(portfolio_returns - benchmark_returns))
total_excess = allocation + selection + interaction
return {
'配置效应': allocation,
'选择效应': selection,
'交互效应': interaction,
'总超额收益': total_excess
}
```
### 组合构建步骤
1. **确定投资目标** — 收益目标、风险承受能力、投资期限
2. **选择资产类别** — 股票（国内/国际/新兴市场）、债券（国债/公司债/高收益债）、另类投资（房地产/大宗商品）
3. **确定资产权重** — 运用优化方法、考虑约束条件
4. **选择具体标的** — ETF vs 个股、主动 vs 被动
5. **设定再平衡规则** — 频率、触发阈值
6. **持续监控和调整** — 定期绩效评估、根据情况调整策略
---
## 策略优化工具
### 参数网格搜索
```python
from itertools import product
def grid_search_optimization(df, strategy_class, param_grid):
"""网格搜索优化"""
results = []
param_names = list(param_grid.keys())
param_values = list(param_grid.values())
for params in product(*param_values):
param_dict = dict(zip(param_names, params))
strategy = strategy_class(**param_dict)
signals = strategy.backtest(df)
returns = signals['strategy_returns'].dropna()
analyzer = PerformanceMetrics(returns)
results.append({
**param_dict,
'sharpe': analyzer.sharpe_ratio(),
'return': analyzer.total_return(),
'max_dd': analyzer.max_drawdown()
})
return pd.DataFrame(results).sort_values('sharpe', ascending=False)
# 使用示例
param_grid = {
'short_window': [5, 10, 15, 20],
'long_window': [20, 30, 50, 60]
}
results = grid_search_optimization(df, DualMovingAverageStrategy, param_grid)
```
### 止损止盈控制
```python
class StrategyWithRiskControl:
"""带风控的策略"""
def __init__(self, base_strategy, stop_loss=0.05, take_profit=0.10):
self.base_strategy = base_strategy
self.stop_loss = stop_loss
self.take_profit = take_profit
def backtest_with_stops(self, df):
"""带止损止盈的回测"""
signals = self.base_strategy.generate_signals(df)
position = 0
entry_price = 0
trades = []
for i in range(1, len(signals)):
current_price = signals['price'].iloc[i]
signal = signals['signal'].iloc[i]
if position == 0:
if signal == 1:
position = 1
entry_price = current_price
trades.append({'type': 'BUY', 'price': current_price, 'date': signals.index[i]})
elif position == 1:
pnl = (current_price - entry_price) / entry_price
if pnl <= -self.stop_loss:
position = 0
trades.append({'type': 'STOP_LOSS', 'price': current_price, 'date': signals.index[i]})
elif pnl >= self.take_profit:
position = 0
trades.append({'type': 'TAKE_PROFIT', 'price': current_price, 'date': signals.index[i]})
elif signal == -1:
position = 0
trades.append({'type': 'SELL', 'price': current_price, 'date': signals.index[i]})
return pd.DataFrame(trades)
```
### 样本外测试
```python
def walk_forward_test(df, strategy_class, params, train_size=252, test_size=63):
"""滚动窗口测试"""
results = []
for start in range(0, len(df) - train_size - test_size, test_size):
train_data = df.iloc[start:start + train_size]
test_data = df.iloc[start + train_size:start + train_size + test_size]
strategy = strategy_class(**params)
test_signals = strategy.backtest(test_data)
test_returns = test_signals['strategy_returns'].dropna()
results.append({
'period_start': test_data.index[0],
'period_end': test_data.index[-1],
'return': (1 + test_returns).prod() - 1
})
return pd.DataFrame(results)
```
---
## 模拟交易与风控系统
### 模拟交易系统
```python
class PaperTradingSystem:
"""模拟交易系统"""
def __init__(self, initial_capital=100000):
self.capital = initial_capital
self.positions = {}
self.trades_history = []
self.portfolio_history = []
def get_portfolio_value(self, current_prices):
"""计算组合价值"""
position_value = sum(
shares * current_prices.get(symbol, 0)
for symbol, shares in self.positions.items()
)
return self.capital + position_value
def execute_order(self, symbol, shares, price, order_type='BUY'):
"""执行订单"""
if order_type == 'BUY':
cost = shares * price * 1.001
if cost <= self.capital:
self.capital -= cost
self.positions[symbol] = self.positions.get(symbol, 0) + shares
self.trades_history.append({
'time': pd.Timestamp.now(),
'symbol': symbol,
'type': 'BUY',
'shares': shares,
'price': price
})
return True
elif order_type == 'SELL':
if self.positions.get(symbol, 0) >= shares:
revenue = shares * price * 0.999
self.capital += revenue
self.positions[symbol] -= shares
self.trades_history.append({
'time': pd.Timestamp.now(),
'symbol': symbol,
'type': 'SELL',
'shares': shares,
'price': price
})
return True
return False
def record_snapshot(self, current_prices):
"""记录组合快照"""
self.portfolio_history.append({
'time': pd.Timestamp.now(),
'portfolio_value': self.get_portfolio_value(current_prices),
'cash': self.capital,
'positions': self.positions.copy()
})
```
### 风险监控系统
```python
class RiskMonitor:
"""风险监控系统"""
def __init__(self, max_drawdown_limit=0.15, daily_var_limit=0.03):
self.max_dd_limit = max_drawdown_limit
self.var_limit = daily_var_limit
self.alerts = []
def check_drawdown(self, portfolio_values):
"""检查回撤"""
peak = max(portfolio_values)
current = portfolio_values[-1]
drawdown = (peak - current) / peak
if drawdown > self.max_dd_limit:
self.alerts.append({
'type': 'DRAWDOWN_ALERT',
'message': f'回撤 {drawdown:.2%} 超过限制 {self.max_dd_limit:.2%}',
'level': 'HIGH'
})
return False
return True
def check_daily_var(self, returns, confidence=0.95):
"""检查日VaR"""
var = np.percentile(returns, (1 - confidence) * 100)
if abs(var) > self.var_limit:
self.alerts.append({
'type': 'VAR_ALERT',
'message': f'VaR {abs(var):.2%} 超过限制 {self.var_limit:.2%}',
'level': 'MEDIUM'
})
return False
return True
def get_alerts(self):
"""获取所有警报"""
return self.alerts
```
---
## 分析报告模板
### 标准股票分析报告结构
```markdown
# [股票代码] 深度分析报告
## 📊 摘要
- **当前价格**: $XXX.XX
- **分析日期**: YYYY-MM-DD
- **综合评级**: ⭐⭐⭐⭐ (X/5)
## 📈 技术面分析
### 趋势判断
- 短期趋势: 上涨/下跌/震荡
- 中期趋势: 上涨/下跌/震荡
- 长期趋势: 上涨/下跌/震荡
### 关键指标
| 指标 | 数值 | 信号 |
|------|------|------|
| RSI(14) | XX | 超买/超卖/中性 |
| MACD | XX | 金叉/死叉/中性 |
| 布林带位置 | XX% | 上轨/中轨/下轨 |
### 支撑与阻力
- 强阻力位: $XXX
- 弱阻力位: $XXX
- 弱支撑位: $XXX
- 强支撑位: $XXX
## 💰 基本面分析
### 估值指标
| 指标 | 公司 | 行业平均 | 评价 |
|------|------|----------|------|
| P/E | XX | XX | 高估/低估/合理 |
| P/B | XX | XX | 高估/低估/合理 |
| ROE | XX% | XX% | 优秀/良好/一般 |
### 财务健康度
- 流动比率: XX
- 资产负债率: XX%
- 利息保障倍数: XX
## ⚠️ 风险评估
| 指标 | 数值 | 风险等级 |
|------|------|----------|
| 年化波动率 | XX% | 低/中/高 |
| 最大回撤 | XX% | 低/中/高 |
| VaR(95%) | XX% | 低/中/高 |
| Beta | XX | 低/中/高 |
## 🎯 投资建议
**操作建议**: 买入/持有/卖出/观望
**建议理由**:
1. ...
2. ...
3. ...
**目标价位**: $XXX - $XXX
**止损价位**: $XXX
---
> **风险提示**：本分析仅供参考，不构成投资建议。金融市场存在风险，投资需谨慎。
```
---
## 量化金融入门与职业指南
### 什么是量化金融？
**量化金融（Quantitative Finance）** 是运用数学模型、统计方法和计算机技术来分析金融市场、管理风险和制定投资决策的学科。
```
量化金融 = 金融理论 + 数学建模 + 统计分析 + 编程技术
```
**与传统金融的区别**：
| 对比维度 | 传统金融 | 量化金融 |
|----------|----------|----------|
| 决策依据 | 经验、直觉、定性分析 | 数据、模型、定量分析 |
| 分析方法 | 基本面研究、专家判断 | 统计建模、机器学习 |
| 交易执行 | 人工交易 | 算法自动化交易 |
| 风险管理 | 主观评估 | 量化风险模型 |
| 可重复性 | 难以复制 | 系统化、可回测 |
### 量化金融的发展历史
| 时期 | 里程碑 | 影响 |
|------|--------|------|
| 1952年 | 马科维茨提出现代投资组合理论 | 风险-收益量化分析的基础 |
| 1973年 | Black-Scholes期权定价公式 | 衍生品定价的数学革命 |
| 1980年代 | 计算机交易兴起 | 算法交易的萌芽 |
| 1990年代 | VaR风险度量普及 | 风险管理标准化 |
| 2000年代 | 高频交易发展 | 微秒级交易竞争 |
| 2010年代 | 机器学习应用 | AI驱动的量化策略 |
| 2020年代 | 另类数据、深度学习 | 数据来源多元化 |
### 应用领域
#### 量化交易策略类型
| 策略类型 | 原理 | 典型周期 |
|----------|------|----------|
| 趋势跟踪 | 追随市场趋势方向 | 日-月 |
| 均值回归 | 价格偏离后会回归均值 | 日-周 |
| 统计套利 | 利用资产间的统计关系 | 分钟-日 |
| 市场中性 | 多空对冲，消除市场风险 | 日-月 |
| 高频交易 | 超短期微小价差获利 | 微秒-秒 |
| 事件驱动 | 利用公司事件（财报、并购）| 日-周 |
#### 其他应用
- **风险管理**：识别、量化和管理各类金融风险
- **衍生品定价**：期权、期货等的公允价值计算
- **资产管理**：投资组合优化、智能Beta、因子投资
- **金融科技**：智能投顾、信用评分、反欺诈
### 从业所需技能
#### 技能金字塔
```
┌─────────────┐
│ 领域专长 │ ← 特定策略/产品深耕
┌─┴─────────────┴─┐
│ 金融知识 │ ← 市场、产品、法规
┌─┴─────────────────┴─┐
│ 量化技能 │ ← 数学、统计、建模
┌─┴─────────────────────┴─┐
│ 编程能力 │ ← Python、C++、SQL
┌─┴─────────────────────────┴─┐
│ 基础素质 │ ← 逻辑思维、学习能力
└─────────────────────────────┘
```
#### 数学基础
| 数学领域 | 重要知识点 | 应用场景 |
|----------|------------|----------|
| **微积分** | 导数、积分、多元函数 | 衍生品定价、优化 |
| **线性代数** | 矩阵运算、特征值、SVD | 投资组合优化、PCA |
| **概率论** | 概率分布、随机变量、期望 | 风险建模、蒙特卡洛 |
| **统计学** | 假设检验、回归分析、时间序列 | 策略回测、因子分析 |
| **随机过程** | 布朗运动、伊藤引理、鞅 | 期权定价、风险模型 |
| **优化理论** | 凸优化、约束优化、梯度下降 | 组合优化、参数调优 |
**学习优先级**：
1. 概率论与统计学 ⭐⭐⭐⭐⭐
2. 线性代数 ⭐⭐⭐⭐⭐
3. 微积分 ⭐⭐⭐⭐
4. 随机过程 ⭐⭐⭐⭐
5. 优化理论 ⭐⭐⭐
#### 编程技能
| 语言 | 用途 | 重要性 |
|------|------|--------|
| **Python** | 数据分析、策略研究、机器学习 | ⭐⭐⭐⭐⭐ |
| **SQL** | 数据库查询、数据提取 | ⭐⭐⭐⭐⭐ |
| **C/C++** | 高频交易、低延迟系统 | ⭐⭐⭐⭐ |
| **R** | 统计分析、学术研究 | ⭐⭐⭐ |
**Python核心库**：
```python
# 数据处理
import pandas as pd # 数据处理
import numpy as np # 数值计算
# 可视化
import matplotlib.pyplot as plt
import seaborn as sns
# 统计建模
import statsmodels.api as sm
from scipy import stats
# 机器学习
from sklearn.ensemble import RandomForestClassifier
from sklearn.linear_model import LinearRegression
# 量化金融专用
import yfinance as yf # 股票数据
import backtrader as bt # 回测框架
import zipline # 回测框架
import pyfolio # 绩效分析
```
#### 机器学习（进阶）
| 算法类型 | 具体方法 | 量化应用 |
|----------|----------|----------|
| 监督学习 | 线性回归、随机森林、XGBoost | 收益预测、分类 |
| 非监督学习 | K-means、PCA、聚类 | 资产分类、降维 |
| 深度学习 | LSTM、Transformer | 序列预测、NLP |
| 强化学习 | Q-learning、DQN | 交易策略优化 |
### 学习路径规划
#### 阶段一：基础构建（3-6个月）
| 学习内容 | 推荐资源 | 预计时间 |
|----------|----------|----------|
| Python编程 | 《Python编程：从入门到实践》 | 1-2个月 |
| 概率统计 | 《概率论与数理统计》(茆诗松) | 2个月 |
| 线性代数 | MIT 18.06 线性代数课程 | 1个月 |
| SQL基础 | LeetCode SQL练习 | 2周 |
#### 阶段二：量化入门（3-6个月）
| 学习内容 | 推荐资源 | 预计时间 |
|----------|----------|----------|
| 金融市场基础 | 《金融市场学》 | 1个月 |
| 投资学 | 《投资学》(博迪) | 2个月 |
| 量化交易入门 | 《打开量化投资的黑箱》 | 1个月 |
| pandas数据分析 | 《利用Python进行数据分析》 | 1个月 |
#### 阶段三：技能深化（6-12个月）
| 学习内容 | 推荐资源 | 预计时间 |
|----------|----------|----------|
| 时间序列分析 | 《时间序列分析》(Hamilton) | 2个月 |
| 机器学习 | 《机器学习实战》、Coursera ML课程 | 3个月 |
| 金融工程 | 《期权、期货及其他衍生产品》(Hull) | 2个月 |
| 风险管理 | 《风险管理与金融机构》 | 1个月 |
#### 阶段四：专业进阶（持续）
| 方向 | 进阶学习 | 适合人群 |
|------|----------|----------|
| 量化研究 | 高级计量经济学、论文阅读 | 偏学术、研究型 |
| 策略开发 | 实盘交易、策略优化 | 偏实战、交易型 |
| 风险管理 | FRM认证、高级风险模型 | 偏稳健、风控型 |
| 系统开发 | C++、系统架构、低延迟 | 偏技术、工程型 |
### 职业发展路径
#### 典型岗位
| 岗位 | 主要职责 | 技能侧重 | 薪资范围(年薪) |
|------|----------|----------|----------------|
| **量化研究员** | 策略研究、因子挖掘 | 数学、统计、建模 | 30-150万+ |
| **量化开发** | 策略实现、系统开发 | 编程、系统设计 | 30-100万+ |
| **量化交易员** | 策略执行、风险监控 | 交易、风控 | 30-200万+ |
| **风险分析师** | 风险建模、压力测试 | 风险模型、监管 | 25-80万 |
| **数据科学家** | 数据挖掘、特征工程 | ML、大数据 | 30-100万 |
#### 职业晋升路径
```
初级量化分析师 (0-2年)
↓
量化分析师/研究员 (2-5年)
↓
高级量化研究员 (5-8年)
↓
┌───┴───┐
↓ ↓
量化总监 首席投资官
(管理路线) (专家路线)
```
#### 主要就业方向
| 机构类型 | 代表公司 | 特点 |
|----------|----------|------|
| **对冲基金** | Two Sigma、Citadel、桥水 | 高薪、高压、纯量化 |
| **自营交易公司** | Jane Street、Optiver、Jump | 技术导向、高频 |
| **投资银行** | 高盛、摩根士丹利、中金 | 综合业务、大平台 |
| **资产管理** | 贝莱德、富达、华夏基金 | 规模大、稳定 |
| **量化私募** | 幻方、九坤、明汯 | 国内头部、机会多 |
| **金融科技** | 蚂蚁、京东数科 | 技术创新、互联网 |
### 面试准备
#### 常见面试题类型
**1. 概率统计题**
```
例：连续抛硬币，直到连续出现两次正面，期望抛多少次？
```
**2. 编程题**
```python
# 例：计算股票的最大回撤
def max_drawdown(prices):
peak = prices[0]
max_dd = 0
for price in prices:
if price > peak:
peak = price
dd = (peak - price) / peak
max_dd = max(max_dd, dd)
return max_dd
```
**3. 金融知识题**
```
例：解释什么是夏普比率？它有什么局限性？
```
**4. 脑筋急转弯**
```
例：一根绳子烧完要1小时（但燃烧速度不均匀），
如何用两根这样的绳子计时45分钟？
```
#### 面试准备清单
- [ ] 熟练掌握Python/SQL编程
- [ ] 复习概率论和统计学
- [ ] 了解常见量化策略原理
- [ ] 准备1-2个个人量化项目
- [ ] 练习LeetCode算法题
- [ ] 阅读经典量化金融书籍
- [ ] 了解目标公司和行业动态
### 推荐资源
#### 必读书籍
| 书名 | 作者 | 适合阶段 |
|------|------|----------|
| 《Python金融大数据分析》 | Yves Hilpisch | 入门 |
| 《打开量化投资的黑箱》 | Rishi K. Narang | 入门 |
| 《主动投资组合管理》 | Grinold & Kahn | 进阶 |
| 《期权、期货及其他衍生产品》 | John Hull | 进阶 |
| 《Advances in Financial ML》 | Marcos López | 高级 |
#### 在线学习平台
- **Coursera**：金融工程、机器学习课程
- **Quantopian**：量化策略开发（历史资源）
- **QuantConnect**：免费回测平台
- **Kaggle**：数据科学竞赛
#### 实践平台
| 平台 | 用途 | 链接 |
|------|------|------|
| JoinQuant | 国内量化回测 | joinquant.com |
| RiceQuant | 国内量化回测 | ricequant.com |
| QuantConnect | 国际量化平台 | quantconnect.com |
| Alpaca | 美股模拟交易 | alpaca.markets |
### 给初学者的建议
**✅ 应该做的**：
1. 从基础开始，先打好数学和编程基础
2. 理论学习后立即动手写代码
3. 完成1-2个完整的量化项目
4. 保持持续学习
5. 技术之外，理解真实的市场运作
**❌ 应该避免的**：
1. 不要跳过基础直接学策略
2. 避免过拟合历史数据
3. 收益重要，但风险管理更重要
4. 多与业内人士交流
5. 不存在永远有效的策略
> "量化金融不是寻找完美策略，而是在不确定性中做出最优决策。"
---
## 量化实操项目指南
系统化的量化金融实操指导，通过分阶段的项目实践帮助用户从零基础到能够独立开发量化策略。
### 实操路线图
```
第1阶段: 数据基础 第2阶段: 指标计算 第3阶段: 策略开发
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ • 数据获取 │ ──▶ │ • 技术指标 │ ──▶ │ • 简单策略 │
│ • 数据清洗 │ │ • 统计分析 │ │ • 回测框架 │
│ • 可视化 │ │ • 因子计算 │ │ • 绩效评估 │
└─────────────┘ └─────────────┘ └─────────────┘
│
▼
第6阶段: 实盘准备 第5阶段: 组合管理 第4阶段: 策略优化
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ • 风控系统 │ ◀── │ • 资产配置 │ ◀── │ • 参数优化 │
│ • 模拟交易 │ │ • 再平衡 │ │ • 多策略 │
│ • 监控报警 │ │ • 归因分析 │ │ • 风险控制 │
└─────────────┘ └─────────────┘ └─────────────┘
```
各阶段的详细练习任务、代码示例和验收标准已包含在上方对应的代码模板库和工作流中。按照路线图逐步完成即可。
### 综合实战项目：构建完整量化交易系统
**项目结构**：
```
quant_system/
├── data/
│ ├── data_fetcher.py # 数据获取
│ └── data_processor.py # 数据处理
├── indicators/
│ ├── technical.py # 技术指标
│ └── fundamental.py # 基本面指标
├── strategies/
│ ├── base_strategy.py # 策略基类
│ ├── ma_strategy.py # 均线策略
│ └── momentum_strategy.py # 动量策略
├── backtest/
│ ├── engine.py # 回测引擎
│ └── analyzer.py # 绩效分析
├── risk/
│ ├── risk_manager.py # 风险管理
│ └── position_sizing.py # 仓位管理
├── portfolio/
│ ├── optimizer.py # 组合优化
│ └── rebalancer.py # 再平衡
└── main.py # 主程序
```
**完成标准**：
- [ ] 能够获取多只股票数据
- [ ] 实现至少2种交易策略
- [ ] 完整的回测和绩效评估
- [ ] 基本的风险控制功能
- [ ] 投资组合优化功能
- [ ] 清晰的代码文档
### 实操检查清单
**初级（完成第1-2阶段后）**
- [ ] 能独立获取和清洗股票数据
- [ ] 能计算常用技术指标（MA、RSI、MACD）
- [ ] 能绑制基本的金融图表
**中级（完成第3-4阶段后）**
- [ ] 能开发简单的量化策略
- [ ] 能进行完整的策略回测
- [ ] 理解并能计算各种绩效指标
- [ ] 能进行参数优化
**高级（完成第5-6阶段后）**
- [ ] 能构建多资产投资组合
- [ ] 能实现组合优化和再平衡
- [ ] 能建立风险监控系统
- [ ] 能搭建模拟交易环境
---
## 文件与输出规范
### 目录结构
| 类型 | 目录 | 命名规则 |
|------|------|----------|
| 数据文件 | `data/` | `{symbol}_price.json` |
| 代码文件 | `code/` | `analysis_{symbol}.py` |
| 图表文件 | `charts/` | `{symbol}_{type}.png` |
| 报告文件 | `reports/` | `{symbol}_report.md` |
### 图表要求
- 所有图表必须有清晰的标题
- 坐标轴必须有标签
- 使用专业的配色方案
- 保存为高质量PNG格式到 `charts/` 目录
### 代码标准结构
```python
"""
文件: code/analysis_{symbol}_{date}.py
功能: {分析描述}
"""
# 1. 导入库
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import json
# 2. 配置
plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False
# 3. 加载数据
# ...
# 4. 分析计算
# ...
# 5. 可视化
# ...
# 6. 保存结果
# ...
# 7. 输出摘要
print("=== 分析完成 ===")
```
---
## 专业术语
确保正确使用以下金融术语：
- **Alpha（阿尔法）**：超额收益
- **Beta（贝塔）**：系统性风险
- **Sharpe Ratio（夏普比率）**：风险调整收益
- **Drawdown（回撤）**：从峰值的下跌幅度
- **Volatility（波动率）**：价格变动的标准差
- **P/E Ratio（市盈率）**：价格与每股收益的比率
- **ROE（净资产收益率）**：净利润与股东权益的比率
---
## 重要声明
在每次提供投资建议时，必须包含以下风险声明：
> **风险提示**：本分析仅供参考，不构成投资建议。金融市场存在风险，投资需谨慎。过往表现不代表未来收益。请根据自身风险承受能力做出投资决策。
Attribution
Comments
Loading comments…