从金融市场分析宏观经济：股票市场作为经济晴雨表

1
2
3
4
5
6

经济周期阶段 → 股市表现

复苏期：股市开始上涨，领先经济复苏
繁荣期：股市持续上涨，与经济同步
衰退期：股市开始下跌，领先经济衰退
萧条期：股市持续下跌，与经济同步

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

# 行业轮动分析
def industry_rotation_analysis(economic_phase):
    """
    根据经济周期阶段分析行业轮动
    economic_phase: 'recovery', 'expansion', 'recession', 'contraction'
    """
    industry_map = {
        'recovery': {
            '领先行业': ['金融', '房地产', '可选消费'],
            '同步行业': ['工业', '材料', '能源'],
            '滞后行业': ['公用事业', '必需消费']
        },
        'expansion': {
            '领先行业': ['科技', '可选消费', '工业'],
            '同步行业': ['材料', '能源', '金融'],
            '滞后行业': ['公用事业', '必需消费']
        },
        'recession': {
            '领先行业': ['必需消费', '公用事业', '医疗'],
            '同步行业': ['金融', '房地产'],
            '滞后行业': ['科技', '可选消费', '工业']
        },
        'contraction': {
            '领先行业': ['必需消费', '公用事业'],
            '同步行业': ['医疗', '金融'],
            '滞后行业': ['科技', '可选消费', '工业', '材料']
        }
    }
    return industry_map.get(economic_phase, {})

# 示例
recovery_industries = industry_rotation_analysis('recovery')
print("复苏期行业轮动：", recovery_industries)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# GDP与股市关系分析
def gdp_stock_analysis(gdp_growth, stock_index_change):
    """
    分析GDP增长与股市变化的关系
    """
    if gdp_growth > 0 and stock_index_change > 0:
        return "经济复苏，股市上涨"
    elif gdp_growth > 0 and stock_index_change < 0:
        return "经济增长但股市下跌，可能存在估值过高或政策风险"
    elif gdp_growth < 0 and stock_index_change < 0:
        return "经济衰退，股市下跌"
    else:
        return "经济衰退但股市上涨，可能存在政策刺激或预期改善"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# 行业相对强度分析
import pandas as pd
import numpy as np

def industry_relative_strength(industry_returns, market_returns, period=20):
    """
    计算行业相对强度
    industry_returns: 行业收益率
    market_returns: 市场收益率
    period: 计算周期
    """
    # 计算累计收益率
    industry_cumulative = (1 + industry_returns).rolling(window=period).apply(np.prod) - 1
    market_cumulative = (1 + market_returns).rolling(window=period).apply(np.prod) - 1

    # 计算相对强度
    relative_strength = industry_cumulative - market_cumulative

    return relative_strength

# 示例
# 假设有行业和市场收益率数据
industry_returns = pd.Series([0.01, 0.02, -0.01, 0.03, 0.01] * 20)
market_returns = pd.Series([0.005, 0.015, -0.005, 0.02, 0.008] * 20)

rs = industry_relative_strength(industry_returns, market_returns)
print("行业相对强度：", rs.tail())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

# 行业轮动识别
def identify_industry_rotation(industry_performance):
    """
    识别行业轮动
    industry_performance: 各行业表现数据
    """
    # 计算各行业相对强度
    relative_strength = {}
    for industry, returns in industry_performance.items():
        rs = returns.mean() / returns.std()  # 简化的相对强度
        relative_strength[industry] = rs

    # 排序找出强势行业
    sorted_industries = sorted(relative_strength.items(),
                              key=lambda x: x[1],
                              reverse=True)

    # 判断经济阶段
    top_industries = [ind[0] for ind in sorted_industries[:3]]

    if '金融' in top_industries or '房地产' in top_industries:
        phase = "复苏期或早期繁荣期"
    elif '科技' in top_industries or '消费' in top_industries:
        phase = "繁荣期"
    elif '必需消费' in top_industries or '公用事业' in top_industries:
        phase = "衰退期或萧条期"
    else:
        phase = "过渡期"

    return {
        '强势行业': top_industries,
        '经济阶段': phase,
        '相对强度': dict(sorted_industries)
    }

# 示例
industry_data = {
    '金融': pd.Series([0.02, 0.03, 0.01, 0.02, 0.015] * 20),
    '科技': pd.Series([0.01, 0.015, -0.01, 0.02, 0.01] * 20),
    '消费': pd.Series([0.015, 0.02, 0.005, 0.015, 0.01] * 20),
    '必需消费': pd.Series([0.005, 0.01, 0.008, 0.006, 0.007] * 20)
}

rotation_info = identify_industry_rotation(industry_data)
print("行业轮动分析：", rotation_info)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

# 行业景气度分析
def industry_prosperity_analysis(industry_data):
    """
    分析行业景气度
    industry_data: 包含各项指标的行业数据
    """
    prosperity_score = 0

    # 新订单指数
    if industry_data['new_orders'] > 50:
        prosperity_score += 2
    elif industry_data['new_orders'] > 45:
        prosperity_score += 1

    # 营业收入增长率
    if industry_data['revenue_growth'] > 0.1:
        prosperity_score += 2
    elif industry_data['revenue_growth'] > 0:
        prosperity_score += 1

    # 利润增长率
    if industry_data['profit_growth'] > 0.15:
        prosperity_score += 2
    elif industry_data['profit_growth'] > 0:
        prosperity_score += 1

    # 股价表现
    if industry_data['stock_return'] > 0.1:
        prosperity_score += 2
    elif industry_data['stock_return'] > 0:
        prosperity_score += 1

    # 判断景气度
    if prosperity_score >= 7:
        prosperity = "高度景气"
    elif prosperity_score >= 5:
        prosperity = "中度景气"
    elif prosperity_score >= 3:
        prosperity = "低度景气"
    else:
        prosperity = "不景气"

    return {
        '景气度得分': prosperity_score,
        '景气度判断': prosperity,
        '建议': "关注" if prosperity_score >= 5 else "谨慎"
    }

# 示例
sample_data = {
    'new_orders': 52,
    'revenue_growth': 0.12,
    'profit_growth': 0.18,
    'stock_return': 0.15
}

prosperity = industry_prosperity_analysis(sample_data)
print("行业景气度：", prosperity)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

# 宏观经济环境综合分析
def comprehensive_macro_analysis(macro_data, industry_data, market_data):
    """
    综合分析宏观经济环境
    """
    analysis_result = {
        '经济阶段': None,
        '主要特征': [],
        '投资建议': [],
        '风险提示': []
    }

    # 1. 判断经济阶段
    gdp_growth = macro_data['gdp_growth']
    pmi = macro_data['pmi']
    cpi = macro_data['cpi']

    if gdp_growth > 0.06 and pmi > 50:
        analysis_result['经济阶段'] = "繁荣期"
        analysis_result['主要特征'].append("经济增长强劲")
        analysis_result['投资建议'].append("关注成长性行业：科技、消费")
    elif gdp_growth > 0.03 and pmi > 50:
        analysis_result['经济阶段'] = "复苏期"
        analysis_result['主要特征'].append("经济温和复苏")
        analysis_result['投资建议'].append("关注周期性行业：金融、房地产")
    elif gdp_growth < 0 or pmi < 50:
        analysis_result['经济阶段'] = "衰退期"
        analysis_result['主要特征'].append("经济下行压力")
        analysis_result['投资建议'].append("关注防御性行业：必需消费、公用事业")

    # 2. 通胀分析
    if cpi > 0.05:
        analysis_result['风险提示'].append("高通胀风险，关注政策收紧")
        analysis_result['投资建议'].append("关注抗通胀资产：资源类、房地产")
    elif cpi < 0:
        analysis_result['风险提示'].append("通缩风险，关注需求不足")
        analysis_result['投资建议'].append("关注必需消费和公用事业")

    # 3. 行业轮动分析
    top_industries = industry_data['top_performers']
    if '金融' in top_industries:
        analysis_result['主要特征'].append("金融板块领涨，流动性宽松")
    if '科技' in top_industries:
        analysis_result['主要特征'].append("科技板块活跃，创新驱动")

    return analysis_result

# 示例
macro_info = {
    'gdp_growth': 0.05,
    'pmi': 51,
    'cpi': 0.02
}

industry_info = {
    'top_performers': ['金融', '科技', '消费']
}

market_info = {
    'market_return': 0.08,
    'volatility': 0.15
}

result = comprehensive_macro_analysis(macro_info, industry_info, market_info)
print("综合分析结果：", result)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

# 使用Tushare获取行业数据
import tushare as ts
import pandas as pd

# 初始化（需要token）
# ts.set_token('your_token')
# pro = ts.pro_api()

def get_industry_data():
    """
    获取行业板块数据
    """
    # 获取行业分类
    # industries = pro.index_classify(level='L1', src='SW2021')

    # 获取行业指数
    # industry_index = pro.index_daily(ts_code='801010.SI')  # 示例：农林牧渔

    # 获取行业成分股
    # stocks = pro.index_weight(index_code='801010.SI')

    pass

# 计算行业表现
def calculate_industry_performance(industry_stocks, start_date, end_date):
    """
    计算行业整体表现
    """
    # 获取各股票价格数据
    # 计算行业平均收益率
    # 计算行业相对强度
    pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

# 行业表现可视化
import matplotlib.pyplot as plt
import seaborn as sns

def plot_industry_performance(industry_data):
    """
    绘制行业表现对比图
    """
    # 准备数据
    industries = list(industry_data.keys())
    returns = [industry_data[ind]['return'] for ind in industries]

    # 创建图表
    plt.figure(figsize=(12, 6))
    bars = plt.barh(industries, returns,
                    color=['green' if r > 0 else 'red' for r in returns])

    plt.xlabel('收益率 (%)')
    plt.title('行业表现对比')
    plt.axvline(x=0, color='black', linestyle='--', linewidth=0.5)
    plt.grid(axis='x', alpha=0.3)

    # 添加数值标签
    for i, (ind, ret) in enumerate(zip(industries, returns)):
        plt.text(ret, i, f'{ret:.2f}%',
                va='center', ha='left' if ret > 0 else 'right')

    plt.tight_layout()
    plt.show()

# 示例数据
sample_industry_data = {
    '金融': {'return': 8.5},
    '科技': {'return': 12.3},
    '消费': {'return': 6.2},
    '医药': {'return': 4.8},
    '能源': {'return': -2.1}
}

# plot_industry_performance(sample_industry_data)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

# 行业轮动热力图
def plot_industry_rotation_heatmap(rotation_data):
    """
    绘制行业轮动热力图
    rotation_data: 时间序列的行业表现数据
    """
    # 转换为DataFrame
    df = pd.DataFrame(rotation_data)

    # 创建热力图
    plt.figure(figsize=(14, 8))
    sns.heatmap(df.T, annot=True, fmt='.2f', cmap='RdYlGn',
                center=0, cbar_kws={'label': '相对强度'})

    plt.title('行业轮动热力图')
    plt.xlabel('时间')
    plt.ylabel('行业')
    plt.tight_layout()
    plt.show()