金
金融数据洞察分析
作者：鹿Sir通用技能v1
金融机构数据全量采集与智能分析工具，支持股票、基金、债券、期货等多品类数据采集，提供统计分析、可视化报告和市场洞察。
下载量
249
点赞
62
价格
免费
技能文档
---
name: financial-data-insights
title: "金融数据洞察分析"
description: 金融机构数据全量采集与智能分析工具，支持股票、基金、债券、期货等多品类数据采集，提供统计分析、可视化报告和市场洞察。
category: "通用技能"
---
# 金融数据洞察分析
面向中国金融市场的全流程数据分析技能，覆盖数据采集、清洗、统计分析、可视化报告生成和市场洞察输出。
## 适用场景
- 股票市场数据采集与行情分析
- 基金产品净值追踪与业绩评估
- 债券市场利率走势与信用分析
- 期货商品行情与持仓分析
- 宏观经济指标采集与解读
- 投资组合绩效归因与风险评估
- 行业板块轮动分析与比较
## 工作流程
### 步骤 1 · 需求确认
首先明确用户的分析需求：
1. **数据类型**：股票 / 基金 / 债券 / 期货 / 宏观数据 / 综合
2. **分析范围**：单标的深度分析 / 多标的横向对比 / 行业板块分析
3. **时间跨度**：日线级别 / 周线级别 / 月线级别 / 自定义区间
4. **输出形式**：统计报告 / 可视化图表 / 综合分析文档
5. **特殊需求**：技术指标计算 / 风险指标评估 / 相关性分析
### 步骤 2 · 数据采集
#### 2.1 数据源选择
根据数据类型选择合适的数据获取方式：
```python
# 常用金融数据接口
import akshare as ak # A股、基金、期货等中国市场数据
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
# 备选数据源
# tushare: 需要token，适合专业量化
# baostock: 免费A股数据
# yfinance: 海外市场数据
```
#### 2.2 股票数据采集
```python
def fetch_stock_data(symbol, start_date, end_date):
"""
采集A股股票历史行情数据
参数:
symbol: 股票代码，如 '000001'（平安银行）
start_date: 开始日期，格式 'YYYYMMDD'
end_date: 结束日期，格式 'YYYYMMDD'
"""
try:
# 使用 akshare 获取日线数据
df = ak.stock_zh_a_hist(
symbol=symbol,
period="daily",
start_date=start_date,
end_date=end_date,
adjust="qfq" # 前复权
)
# 标准化列名
df.columns = ['日期', '开盘价', '收盘价', '最高价', '最低价',
'成交量', '成交额', '振幅', '涨跌幅', '涨跌额', '换手率']
df['日期'] = pd.to_datetime(df['日期'])
df.set_index('日期', inplace=True)
print(f"成功获取 {symbol} 数据，共 {len(df)} 条记录")
print(f"时间范围: {df.index[0].strftime('%Y-%m-%d')} 至 {df.index[-1].strftime('%Y-%m-%d')}")
return df
except Exception as e:
print(f"数据采集失败: {e}")
return None
# 采集示例
# stock_df = fetch_stock_data('000001', '20240101', '20241231')
```
#### 2.3 基金数据采集
```python
def fetch_fund_data(fund_code, start_date, end_date):
"""
采集基金净值数据
参数:
fund_code: 基金代码，如 '110011'
start_date: 开始日期
end_date: 结束日期
"""
try:
# 开放式基金净值
df = ak.fund_open_fund_info_em(symbol=fund_code, indicator="单位净值走势")
df.columns = ['日期', '单位净值', '日增长率']
df['日期'] = pd.to_datetime(df['日期'])
# 按时间范围筛选
mask = (df['日期'] >= start_date) & (df['日期'] <= end_date)
df = df[mask].copy()
df.set_index('日期', inplace=True)
# 获取基金基本信息
fund_info = ak.fund_individual_basic_info_xq(symbol=fund_code)
print(f"成功获取基金 {fund_code} 数据，共 {len(df)} 条记录")
return df, fund_info
except Exception as e:
print(f"基金数据采集失败: {e}")
return None, None
```
#### 2.4 宏观经济数据采集
```python
def fetch_macro_data():
"""
采集中国宏观经济核心指标
"""
macro_data = {}
try:
# GDP 数据
gdp = ak.macro_china_gdp()
macro_data['gdp'] = gdp
# CPI 数据
cpi = ak.macro_china_cpi_monthly()
macro_data['cpi'] = cpi
# PMI 数据
pmi = ak.macro_china_pmi()
macro_data['pmi'] = pmi
# 社会融资规模
sf = ak.macro_china_shrzgm()
macro_data['social_financing'] = sf
# M2 货币供应
m2 = ak.macro_china_m2_year()
macro_data['m2'] = m2
print("宏观经济数据采集完成")
return macro_data
except Exception as e:
print(f"宏观数据采集部分失败: {e}")
return macro_data
```
#### 2.5 行业板块数据采集
```python
def fetch_sector_data():
"""
采集行业板块行情数据
"""
try:
# 申万行业分类行情
sector_df = ak.stock_board_industry_name_em()
# 获取各行业详细数据
sector_details = []
for _, row in sector_df.iterrows():
board_name = row['板块名称']
try:
detail = ak.stock_board_industry_hist_em(
symbol=board_name,
period="日",
start_date="20240101",
end_date="20241231",
adjust=""
)
sector_details.append({
'name': board_name,
'data': detail
})
except:
continue
print(f"成功获取 {len(sector_details)} 个行业板块数据")
return sector_details
except Exception as e:
print(f"行业数据采集失败: {e}")
return []
```
### 步骤 3 · 数据清洗与预处理
```python
def clean_financial_data(df):
"""
金融数据清洗与预处理
"""
# 1. 处理缺失值
print(f"原始数据形状: {df.shape}")
print(f"缺失值统计:\n{df.isnull().sum()}")
# 删除全为空的列
df = df.dropna(axis=1, how='all')
# 前向填充价格类缺失（交易日缺失）
price_cols = [c for c in df.columns if '价' in c or '净值' in c]
for col in price_cols:
df[col] = df[col].ffill()
# 成交量缺失填充为0
vol_cols = [c for c in df.columns if '量' in c or '额' in c]
for col in vol_cols:
df[col] = df[col].fillna(0)
# 2. 异常值检测与处理
for col in price_cols:
q1 = df[col].quantile(0.01)
q99 = df[col].quantile(0.99)
outlier_mask = (df[col] < q1) | (df[col] > q99)
if outlier_mask.sum() > 0:
print(f"{col} 检测到 {outlier_mask.sum()} 个异常值，已进行截断处理")
df.loc[outlier_mask, col] = df.loc[outlier_mask, col].clip(q1, q99)
# 3. 数据类型转换
numeric_cols = df.select_dtypes(include=['object']).columns
for col in numeric_cols:
try:
df[col] = pd.to_numeric(df[col], errors='coerce')
except:
pass
# 4. 时间索引处理
if not isinstance(df.index, pd.DatetimeIndex):
date_col = [c for c in df.columns if '日期' in c or 'date' in c.lower()]
if date_col:
df.index = pd.to_datetime(df[date_col[0]])
df = df.drop(columns=date_col)
df = df.sort_index()
print(f"清洗后数据形状: {df.shape}")
return df
```
### 步骤 4 · 统计分析
#### 4.1 基础统计指标
```python
def compute_statistics(df):
"""
计算金融数据核心统计指标
"""
stats = {}
if '收盘价' in df.columns or '单位净值' in df.columns:
price_col = '收盘价' if '收盘价' in df.columns else '单位净值'
returns = df[price_col].pct_change().dropna()
# 收益率统计
stats['累计收益率'] = (df[price_col].iloc[-1] / df[price_col].iloc[0] - 1) * 100
stats['年化收益率'] = ((1 + stats['累计收益率']/100) ** (252/len(returns)) - 1) * 100
stats['日均收益率'] = returns.mean() * 100
stats['日收益率标准差'] = returns.std() * 100
# 风险指标
stats['最大回撤'] = ((df[price_col] / df[price_col].cummax() - 1).min()) * 100
stats['年化波动率'] = returns.std() * np.sqrt(252) * 100
# 夏普比率（假设无风险利率2%）
rf = 0.02
stats['夏普比率'] = (stats['年化收益率']/100 - rf) / stats['年化波动率'] * 100 if stats['年化波动率'] > 0 else 0
# 偏度和峰度
stats['收益率偏度'] = returns.skew()
stats['收益率峰度'] = returns.kurtosis()
# 涨跌统计
stats['上涨天数'] = (returns > 0).sum()
stats['下跌天数'] = (returns < 0).sum()
stats['平盘天数'] = (returns == 0).sum()
stats['上涨占比'] = stats['上涨天数'] / len(returns) * 100
# 最大单日涨幅/跌幅
stats['最大单日涨幅'] = returns.max() * 100
stats['最大单日跌幅'] = returns.min() * 100
if '涨跌幅' in df.columns:
stats['平均涨跌幅'] = df['涨跌幅'].mean()
stats['涨跌幅标准差'] = df['涨跌幅'].std()
if '换手率' in df.columns:
stats['平均换手率'] = df['换手率'].mean()
stats['最大换手率'] = df['换手率'].max()
if '成交额' in df.columns:
stats['总成交额(亿)'] = df['成交额'].sum() / 1e8
stats['日均成交额(亿)'] = df['成交额'].mean() / 1e8
return stats
def print_statistics(stats):
"""格式化输出统计结果"""
print("\n" + "=" * 50)
print(" 核心统计指标")
print("=" * 50)
for key, value in stats.items():
if isinstance(value, float):
print(f" {key: <12s}: {value:>12.4f}")
else:
print(f" {key: <12s}: {value:>12}")
print("=" * 50)
```
#### 4.2 技术指标计算
```python
def compute_technical_indicators(df):
"""
计算常用技术分析指标
"""
if '收盘价' not in df.columns:
return df
close = df['收盘价']
# 移动平均线
df['MA5'] = close.rolling(5).mean()
df['MA10'] = close.rolling(10).mean()
df['MA20'] = close.rolling(20).mean()
df['MA60'] = close.rolling(60).mean()
# 指数移动平均
df['EMA12'] = close.ewm(span=12).mean()
df['EMA26'] = close.ewm(span=26).mean()
# MACD
df['DIF'] = df['EMA12'] - df['EMA26']
df['DEA'] = df['DIF'].ewm(span=9).mean()
df['MACD'] = 2 * (df['DIF'] - df['DEA'])
# RSI
delta = 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))
# 布林带
df['BOLL_MID'] = close.rolling(20).mean()
std = close.rolling(20).std()
df['BOLL_UP'] = df['BOLL_MID'] + 2 * std
df['BOLL_DN'] = df['BOLL_MID'] - 2 * std
# ATR（平均真实波幅）
high = df['最高价']
low = df['最低价']
prev_close = close.shift(1)
tr = pd.concat([
high - low,
(high - prev_close).abs(),
(low - prev_close).abs()
], axis=1).max(axis=1)
df['ATR'] = tr.rolling(14).mean()
# KDJ
low_14 = low.rolling(14).min()
high_14 = high.rolling(14).max()
rsv = (close - low_14) / (high_14 - low_14) * 100
df['K'] = rsv.ewm(com=2).mean()
df['D'] = df['K'].ewm(com=2).mean()
df['J'] = 3 * df['K'] - 2 * df['D']
print("技术指标计算完成")
return df
```
#### 4.3 相关性分析
```python
def correlation_analysis(stock_list, start_date, end_date):
"""
多股票相关性矩阵分析
参数:
stock_list: 股票代码列表
start_date: 开始日期
end_date: 结束日期
"""
returns_dict = {}
for symbol in stock_list:
df = fetch_stock_data(symbol, start_date, end_date)
if df is not None and '收盘价' in df.columns:
returns_dict[symbol] = df['收盘价'].pct_change()
if not returns_dict:
print("无有效数据用于相关性分析")
return None
# 构建收益率矩阵
returns_df = pd.DataFrame(returns_dict).dropna()
# 计算相关矩阵
corr_matrix = returns_df.corr()
print("\n相关性矩阵:")
print(corr_matrix.round(3))
return corr_matrix, returns_df
```
### 步骤 5 · 可视化报告
#### 5.1 行情走势图
```python
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
# 中文字体配置
plt.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei', 'Noto Sans SC']
plt.rcParams['axes.unicode_minus'] = False
COLORS = ['#1f77b4', '#ff7f0e', '#2ca02c', '#d62728', '#9467bd',
'#8c564b', '#e377c2', '#7f7f7f', '#bcbd22', '#17becf']
def plot_price_chart(df, title='行情走势图', save_path='price_chart.png'):
"""
绘制价格走势 + 成交量双轴图
"""
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(14, 8),
gridspec_kw={'height_ratios': [3, 1]},
sharex=True)
# 价格线
price_col = '收盘价' if '收盘价' in df.columns else '单位净值'
ax1.plot(df.index, df[price_col], color=COLORS[0], linewidth=1.5, label=price_col)
# 均线（如有）
if 'MA20' in df.columns:
ax1.plot(df.index, df['MA20'], color=COLORS[1], linewidth=1,
linestyle='--', alpha=0.7, label='MA20')
if 'MA60' in df.columns:
ax1.plot(df.index, df['MA60'], color=COLORS[2], linewidth=1,
linestyle='--', alpha=0.7, label='MA60')
ax1.set_title(title, fontsize=16, fontweight='bold')
ax1.set_ylabel('价格', fontweight='bold')
ax1.legend(loc='upper left')
ax1.grid(True, alpha=0.3, linestyle='--')
# 成交量柱状图
if '成交额' in df.columns:
colors_bar = [COLORS[3] if df['涨跌幅'].iloc[i] >= 0 else COLORS[0]
for i in range(len(df))]
ax2.bar(df.index, df['成交额'] / 1e8, color=colors_bar, alpha=0.7)
ax2.set_ylabel('成交额(亿)', fontweight='bold')
elif '成交量' in df.columns:
ax2.bar(df.index, df['成交量'] / 1e4, color=COLORS[1], alpha=0.7)
ax2.set_ylabel('成交量(万手)', fontweight='bold')
ax2.set_xlabel('日期', fontweight='bold')
ax2.grid(True, alpha=0.3, linestyle='--')
ax2.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m'))
ax2.xaxis.set_major_locator(mdates.MonthLocator(interval=2))
plt.xticks(rotation=45, ha='right')
plt.tight_layout()
plt.savefig(save_path, dpi=300, bbox_inches='tight', facecolor='white')
plt.close()
print(f"行情走势图已保存: {save_path}")
```
#### 5.2 收益率分布图
```python
def plot_return_distribution(df, save_path='return_dist.png'):
"""
绘制收益率分布直方图 + 核密度估计
"""
price_col = '收盘价' if '收盘价' in df.columns else '单位净值'
returns = df[price_col].pct_change().dropna() * 100
fig, axes = plt.subplots(1, 2, figsize=(14, 5))
# 左图：收益率直方图
from scipy.stats import gaussian_kde, norm
axes[0].hist(returns, bins=50, color=COLORS[0], alpha=0.7,
edgecolor='black', linewidth=0.5, density=True, label='实际分布')
# 核密度估计
kde = gaussian_kde(returns)
x_range = np.linspace(returns.min(), returns.max(), 200)
axes[0].plot(x_range, kde(x_range), color=COLORS[1], linewidth=2, label='KDE')
# 正态分布拟合
mu, std = norm.fit(returns)
x_norm = np.linspace(returns.min(), returns.max(), 200)
axes[0].plot(x_norm, norm.pdf(x_norm, mu, std), color=COLORS[3],
linewidth=2, linestyle='--', label='正态拟合')
axes[0].set_title('收益率分布', fontsize=14, fontweight='bold')
axes[0].set_xlabel('日收益率(%)', fontweight='bold')
axes[0].set_ylabel('密度', fontweight='bold')
axes[0].legend()
axes[0].grid(axis='y', alpha=0.3, linestyle='--')
# 右图：累计收益率曲线
cum_returns = (1 + returns/100).cumprod() - 1
axes[1].plot(df.index[1:], cum_returns * 100, color=COLORS[0], linewidth=1.5)
axes[1].fill_between(df.index[1:], cum_returns * 100, alpha=0.2, color=COLORS[0])
axes[1].axhline(y=0, color='black', linewidth=0.5, linestyle='-')
axes[1].set_title('累计收益率', fontsize=14, fontweight='bold')
axes[1].set_xlabel('日期', fontweight='bold')
axes[1].set_ylabel('累计收益率(%)', fontweight='bold')
axes[1].grid(True, alpha=0.3, linestyle='--')
plt.tight_layout()
plt.savefig(save_path, dpi=300, bbox_inches='tight', facecolor='white')
plt.close()
print(f"收益率分布图已保存: {save_path}")
```
#### 5.3 相关性热力图
```python
def plot_correlation_heatmap(corr_matrix, save_path='correlation_heatmap.png'):
"""
绘制相关性热力图
"""
import seaborn as sns
fig, ax = plt.subplots(figsize=(10, 8))
mask = np.triu(np.ones_like(corr_matrix, dtype=bool))
sns.heatmap(corr_matrix, mask=mask, annot=True, fmt='.2f',
cmap='RdBu_r', center=0, square=True,
linewidths=1, vmin=-1, vmax=1,
cbar_kws={'label': '相关系数'}, ax=ax)
ax.set_title('资产相关性矩阵', fontsize=16, fontweight='bold')
plt.tight_layout()
plt.savefig(save_path, dpi=300, bbox_inches='tight', facecolor='white')
plt.close()
print(f"相关性热力图已保存: {save_path}")
```
#### 5.4 多面板综合分析报告
```python
def plot_comprehensive_report(df, title='综合分析报告', save_path='comprehensive_report.png'):
"""
生成多面板综合分析图
"""
fig = plt.figure(figsize=(18, 14))
price_col = '收盘价' if '收盘价' in df.columns else '单位净值'
returns = df[price_col].pct_change().dropna() * 100
# 1. 价格走势 + 布林带
ax1 = fig.add_subplot(3, 2, 1)
ax1.plot(df.index, df[price_col], color=COLORS[0], linewidth=1.2)
if 'BOLL_UP' in df.columns:
ax1.plot(df.index, df['BOLL_UP'], color=COLORS[1], linewidth=0.8,
linestyle='--', alpha=0.6, label='上轨')
ax1.plot(df.index, df['BOLL_MID'], color=COLORS[2], linewidth=0.8,
linestyle='-', alpha=0.6, label='中轨')
ax1.plot(df.index, df['BOLL_DN'], color=COLORS[3], linewidth=0.8,
linestyle='--', alpha=0.6, label='下轨')
ax1.legend(fontsize=8)
ax1.set_title('价格走势', fontweight='bold')
ax1.grid(True, alpha=0.3, linestyle='--')
# 2. 成交量
ax2 = fig.add_subplot(3, 2, 2)
if '成交额' in df.columns:
colors_bar = [COLORS[3] if df['涨跌幅'].iloc[i] >= 0 else COLORS[0]
for i in range(len(df))]
ax2.bar(df.index, df['成交额'] / 1e8, color=colors_bar, alpha=0.7)
ax2.set_title('成交额(亿)', fontweight='bold')
ax2.grid(True, alpha=0.3, linestyle='--')
# 3. MACD
ax3 = fig.add_subplot(3, 2, 3)
if 'MACD' in df.columns:
colors_macd = [COLORS[3] if v >= 0 else COLORS[0] for v in df['MACD']]
ax3.bar(df.index, df['MACD'], color=colors_macd, alpha=0.6)
ax3.plot(df.index, df['DIF'], color=COLORS[1], linewidth=1, label='DIF')
ax3.plot(df.index, df['DEA'], color=COLORS[2], linewidth=1, label='DEA')
ax3.axhline(y=0, color='black', linewidth=0.5)
ax3.set_title('MACD', fontweight='bold')
ax3.legend(fontsize=8)
ax3.grid(True, alpha=0.3, linestyle='--')
# 4. RSI
ax4 = fig.add_subplot(3, 2, 4)
if 'RSI' in df.columns:
ax4.plot(df.index, df['RSI'], color=COLORS[4], linewidth=1.2)
ax4.axhline(y=70, color=COLORS[3], linewidth=0.8, linestyle='--', label='超买(70)')
ax4.axhline(y=30, color=COLORS[0], linewidth=0.8, linestyle='--', label='超卖(30)')
ax4.set_ylim(0, 100)
ax4.set_title('RSI(14)', fontweight='bold')
ax4.legend(fontsize=8)
ax4.grid(True, alpha=0.3, linestyle='--')
# 5. 收益率分布
ax5 = fig.add_subplot(3, 2, 5)
ax5.hist(returns, bins=40, color=COLORS[0], alpha=0.7,
edgecolor='black', linewidth=0.3, density=True)
ax5.set_title('收益率分布', fontweight='bold')
ax5.set_xlabel('日收益率(%)')
ax5.grid(axis='y', alpha=0.3, linestyle='--')
# 6. 回撤曲线
ax6 = fig.add_subplot(3, 2, 6)
cummax = df[price_col].cummax()
drawdown = (df[price_col] / cummax - 1) * 100
ax6.fill_between(df.index, drawdown, alpha=0.4, color=COLORS[3])
ax6.plot(df.index, drawdown, color=COLORS[3], linewidth=0.8)
ax6.set_title('回撤(%)', fontweight='bold')
ax6.grid(True, alpha=0.3, linestyle='--')
fig.suptitle(title, fontsize=18, fontweight='bold', y=1.01)
plt.tight_layout()
plt.savefig(save_path, dpi=300, bbox_inches='tight', facecolor='white')
plt.close()
print(f"综合分析报告已保存: {save_path}")
```
### 步骤 6 · 市场洞察生成
```python
def generate_market_insights(df, stats):
"""
基于统计数据生成市场洞察文本
"""
insights = []
# 趋势判断
if 'MA20' in df.columns and 'MA60' in df.columns:
latest = df.iloc[-1]
if latest['MA20'] > latest['MA60']:
insights.append("【趋势】短中期均线多头排列，中期趋势偏多")
else:
insights.append("【趋势】短中期均线空头排列，中期趋势偏空")
# 波动率评估
if '年化波动率' in stats:
vol = stats['年化波动率']
if vol < 15:
insights.append(f"【波动】年化波动率 {vol:.1f}%，处于低波动区间，市场相对平稳")
elif vol < 30:
insights.append(f"【波动】年化波动率 {vol:.1f}%，处于正常波动区间")
else:
insights.append(f"【波动】年化波动率 {vol:.1f}%，处于高波动区间，需关注风险")
# 风险指标
if '最大回撤' in stats:
dd = stats['最大回撤']
if dd > -10:
insights.append(f"【风险】最大回撤 {dd:.1f}%，回撤控制良好")
elif dd > -30:
insights.append(f"【风险】最大回撤 {dd:.1f}%，回撤处于中等水平")
else:
insights.append(f"【风险】最大回撤 {dd:.1f}%，回撤较大，需注意风险控制")
# 风险收益比
if '夏普比率' in stats:
sr = stats['夏普比率']
if sr > 1.5:
insights.append(f"【绩效】夏普比率 {sr:.2f}，风险调整后收益优秀")
elif sr > 0.5:
insights.append(f"【绩效】夏普比率 {sr:.2f}，风险调整后收益良好")
else:
insights.append(f"【绩效】夏普比率 {sr:.2f}，风险调整后收益一般")
# 技术面信号
if 'RSI' in df.columns:
rsi = df['RSI'].iloc[-1]
if rsi > 70:
insights.append(f"【技术】RSI={rsi:.1f}，进入超买区间，短期或有回调压力")
elif rsi < 30:
insights.append(f"【技术】RSI={rsi:.1f}，进入超卖区间，可能存在反弹机会")
else:
insights.append(f"【技术】RSI={rsi:.1f}，处于中性区间")
# 成交量分析
if '成交额' in df.columns:
recent_vol = df['成交额'].tail(5).mean()
avg_vol = df['成交额'].mean()
vol_ratio = recent_vol / avg_vol
if vol_ratio > 1.5:
insights.append(f"【量能】近5日成交额为平均水平的 {vol_ratio:.1f} 倍，成交明显放量")
elif vol_ratio < 0.5:
insights.append(f"【量能】近5日成交额仅为平均水平的 {vol_ratio:.1f} 倍，成交明显缩量")
return insights
def print_insights(insights):
"""格式化输出市场洞察"""
print("\n" + "=" * 50)
print(" 市场洞察")
print("=" * 50)
for insight in insights:
print(f" {insight}")
print("=" * 50)
```
### 步骤 7 · 报告输出
```python
def generate_report(symbol, df, stats, insights, output_dir='.'):
"""
生成综合分析报告
"""
import json
price_col = '收盘价' if '收盘价' in df.columns else '单位净值'
report = {
'报告标题': f'{symbol} 金融数据分析报告',
'生成时间': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'数据范围': f"{df.index[0].strftime('%Y-%m-%d')} 至 {df.index[-1].strftime('%Y-%m-%d')}",
'数据条数': len(df),
'最新价格': float(df[price_col].iloc[-1]),
'核心指标': {k: round(v, 4) if isinstance(v, float) else v
for k, v in stats.items()},
'市场洞察': insights
}
# 保存 JSON 报告
report_path = os.path.join(output_dir, f'{symbol}_report.json')
with open(report_path, 'w', encoding='utf-8') as f:
json.dump(report, f, ensure_ascii=False, indent=2)
# 保存 Excel 报告
excel_path = os.path.join(output_dir, f'{symbol}_data.xlsx')
with pd.ExcelWriter(excel_path, engine='openpyxl') as writer:
df.to_excel(writer, sheet_name='行情数据')
# 统计指标表
stats_df = pd.DataFrame(list(stats.items()), columns=['指标', '数值'])
stats_df.to_excel(writer, sheet_name='统计指标', index=False)
# 市场洞察表
insights_df = pd.DataFrame(insights, columns=['洞察内容'])
insights_df.to_excel(writer, sheet_name='市场洞察', index=False)
print(f"\n报告已生成:")
print(f" JSON 报告: {report_path}")
print(f" Excel 报告: {excel_path}")
return report, report_path, excel_path
```
## 完整使用示例
```python
# 完整分析流程
import os
# 1. 采集数据
symbol = '000001'
df = fetch_stock_data(symbol, '20240101', '20241231')
if df is not None:
# 2. 数据清洗
df = clean_financial_data(df)
# 3. 计算技术指标
df = compute_technical_indicators(df)
# 4. 统计分析
stats = compute_statistics(df)
print_statistics(stats)
# 5. 生成可视化
output_dir = './output'
os.makedirs(output_dir, exist_ok=True)
plot_price_chart(df, title=f'{symbol} 行情走势',
save_path=os.path.join(output_dir, 'price_chart.png'))
plot_return_distribution(df,
save_path=os.path.join(output_dir, 'return_dist.png'))
plot_comprehensive_report(df, title=f'{symbol} 综合分析',
save_path=os.path.join(output_dir, 'comprehensive.png'))
# 6. 生成市场洞察
insights = generate_market_insights(df, stats)
print_insights(insights)
# 7. 输出报告
report, json_path, excel_path = generate_report(
symbol, df, stats, insights, output_dir
)
```
## 注意事项
1. **数据源稳定性**：金融数据接口可能因网络或政策原因不稳定，建议设置重试机制
2. **合规要求**：数据分析结果仅供参考，不构成投资建议
3. **时区处理**：A股交易时间为北京时间（UTC+8），注意时区一致性
4. **复权处理**：进行技术分析时使用前复权数据，计算收益率时根据需求选择
5. **大数据量处理**：对于高频数据或长时间跨度，注意内存管理，可分批处理
6. **中文字体**：可视化前确保系统已安装中文字体（SimHei/Microsoft YaHei/Noto Sans SC）
7. **异常数据**：节假日、停牌等特殊交易日的数据需要特别处理支持平台：Qoder · QoderWork · Claude · Codex 等 AI 编程助手