跳到主要内容

经济指标

本页面提供了使用 obb.economy.indicators 端点的入门教程，配合 openbb-econdb 和 openbb-imf 提供商扩展。

国际货币基金组织 (IMF)​

目前有两个 IMF 数据库可通过此命令访问，第三个（贸易方向）可通过 obb.economy.direction_of_trade 获得。

  • 国际储备和外汇流动性
  • 金融稳健性指标
IMF 指标

obb.economy.indicators 端点提供对超过2600个 IMF 时间序列的直接访问，允许选择多个序列和国家。

可用符号​

完整的个别序列列表由 available_indicators 端点返回。

它允许文本查询，其中 ";" 是 AND 操作符，"|" 是 OR 操作符。

# 搜索黄金相关指标
gold_indicators = obb.economy.available_indicators(provider="imf", query="gold")
print(gold_indicators.to_dataframe())
结果
符号符号根描述数据集表格级别顺序子项单位
RAFA_USDRAF_USD官方储备资产，美元IRFCLreserve_assets_and_other_fx_assets22RAFAFX_USD,RAFAIMF_USD,RAFASDR_USD,RAFAGOLD_USD,RAFAO_USDusd
RAFAGOLD_USDRAFA_USD官方储备资产，黄金（包括黄金存款和适当的黄金掉期），美元IRFCLreserve_assets_and_other_fx_assets314RAFAGOLDV_OZTusd

EconDB 提供商​

EconDB 提供商提供对100多个标准化经济指标的访问，覆盖全球主要经济体。

基本用法​

# 获取美国 CPI 数据
us_cpi = obb.economy.indicators(
    symbol="CPI", 
    country="US", 
    provider="econdb"
)
df = us_cpi.to_dataframe()
print(df.head())

常用经济指标​

# 主要经济指标列表
indicators = {
    'CPI': '消费者价格指数',
    'CORE': '核心通胀率',
    'GDP': '国内生产总值',
    'URATE': '失业率',
    'RETAIL': '零售销售',
    'IP': '工业生产',
    'CONF': '消费者信心指数'
}

# 获取美国多个指标
for symbol, name in indicators.items():
    try:
        data = obb.economy.indicators(
            symbol=symbol, 
            country="US", 
            provider="econdb"
        )
        df = data.to_dataframe()
        if not df.empty:
            latest_value = df.iloc[-1]['value']
            latest_date = df.iloc[-1]['date']
            print(f"{name} ({symbol}): {latest_value} ({latest_date})")
    except Exception as e:
        print(f"获取 {symbol} 数据失败: {e}")

多国比较​

# 比较多个国家的 GDP 增长率
countries = ['US', 'CN', 'JP', 'DE', 'GB']
gdp_data = {}

for country in countries:
    try:
        data = obb.economy.indicators(
            symbol="GDP", 
            country=country, 
            provider="econdb"
        )
        df = data.to_dataframe()
        if not df.empty:
            gdp_data[country] = df
            print(f"{country}: 获取到 {len(df)} 条 GDP 记录")
    except Exception as e:
        print(f"获取 {country} GDP 数据失败: {e}")

# 合并数据进行比较
import pandas as pd

if gdp_data:
    combined_df = pd.DataFrame()
    for country, df in gdp_data.items():
        df_country = df.copy()
        df_country['country'] = country
        combined_df = pd.concat([combined_df, df_country], ignore_index=True)
    
    # 显示最新数据
    latest_data = combined_df.groupby('country').last()
    print("\n各国最新 GDP 数据:")
    print(latest_data[['date', 'value']].sort_values('value', ascending=False))

时间序列分析​

def analyze_economic_indicator(symbol, country, years=5):
    """
    分析经济指标的时间序列数据
    """
    from datetime import datetime, timedelta
    
    # 计算日期范围
    end_date = datetime.now()
    start_date = end_date - timedelta(days=years*365)
    
    try:
        data = obb.economy.indicators(
            symbol=symbol,
            country=country,
            start_date=start_date.strftime("%Y-%m-%d"),
            end_date=end_date.strftime("%Y-%m-%d"),
            provider="econdb"
        )
        
        df = data.to_dataframe()
        
        if df.empty:
            print(f"未找到 {country} 的 {symbol} 数据")
            return None
        
        # 基本统计
        stats = {
            'symbol': symbol,
            'country': country,
            'records': len(df),
            'start_date': df['date'].min(),
            'end_date': df['date'].max(),
            'latest_value': df.iloc[-1]['value'],
            'mean_value': df['value'].mean(),
            'std_value': df['value'].std(),
            'min_value': df['value'].min(),
            'max_value': df['value'].max()
        }
        
        return df, stats
        
    except Exception as e:
        print(f"分析 {country} 的 {symbol} 失败: {e}")
        return None, None

# 使用示例
df, stats = analyze_economic_indicator("CPI", "US", years=10)
if stats:
    print("美国 CPI 分析结果:")
    for key, value in stats.items():
        print(f"  {key}: {value}")

通胀率计算​

def calculate_inflation_rate(country, years=5):
    """
    计算通胀率
    """
    try:
        data = obb.economy.indicators(
            symbol="CPI",
            country=country,
            provider="econdb"
        )
        
        df = data.to_dataframe()
        
        if len(df) < 12:
            print(f"数据不足，无法计算 {country} 的通胀率")
            return None
        
        # 按日期排序
        df = df.sort_values('date')
        
        # 计算同比通胀率
        df['inflation_rate'] = df['value'].pct_change(periods=12) * 100
        
        # 获取最新通胀率
        latest_inflation = df.iloc[-1]['inflation_rate']
        latest_date = df.iloc[-1]['date']
        
        print(f"{country} 最新通胀率: {latest_inflation:.2f}% ({latest_date})")
        
        return df[['date', 'value', 'inflation_rate']].dropna()
        
    except Exception as e:
        print(f"计算 {country} 通胀率失败: {e}")
        return None

# 计算多个国家的通胀率
countries = ['US', 'GB', 'DE', 'JP']
inflation_data = {}

for country in countries:
    inflation_df = calculate_inflation_rate(country)
    if inflation_df is not None:
        inflation_data[country] = inflation_df

数据可视化准备​

def prepare_indicator_chart_data(symbol, countries, years=5):
    """
    准备经济指标图表数据
    """
    from datetime import datetime, timedelta
    import pandas as pd
    
    end_date = datetime.now()
    start_date = end_date - timedelta(days=years*365)
    
    chart_data = pd.DataFrame()
    
    for country in countries:
        try:
            data = obb.economy.indicators(
                symbol=symbol,
                country=country,
                start_date=start_date.strftime("%Y-%m-%d"),
                provider="econdb"
            )
            
            df = data.to_dataframe()
            
            if not df.empty:
                df['country'] = country
                chart_data = pd.concat([chart_data, df], ignore_index=True)
                
        except Exception as e:
            print(f"获取 {country} 的 {symbol} 数据失败: {e}")
    
    return chart_data

# 准备多国 GDP 对比数据
gdp_chart_data = prepare_indicator_chart_data(
    "GDP", 
    ["US", "CN", "JP", "DE"], 
    years=10
)

if not gdp_chart_data.empty:
    print("GDP 对比数据准备完成:")
    print(f"总记录数: {len(gdp_chart_data)}")
    print(f"国家数: {gdp_chart_data['country'].nunique()}")
    print(f"日期范围: {gdp_chart_data['date'].min()} 到 {gdp_chart_data['date'].max()}")

经济指标监控​

def monitor_economic_indicators(countries, indicators):
    """
    监控多个国家的经济指标
    """
    results = []
    
    for country in countries:
        country_data = {'country': country}
        
        for symbol, name in indicators.items():
            try:
                data = obb.economy.indicators(
                    symbol=symbol,
                    country=country,
                    provider="econdb"
                )
                
                df = data.to_dataframe()
                
                if not df.empty:
                    latest = df.iloc[-1]
                    country_data[f'{symbol}_value'] = latest['value']
                    country_data[f'{symbol}_date'] = latest['date']
                else:
                    country_data[f'{symbol}_value'] = None
                    country_data[f'{symbol}_date'] = None
                    
            except Exception as e:
                print(f"获取 {country} 的 {symbol} 失败: {e}")
                country_data[f'{symbol}_value'] = None
                country_data[f'{symbol}_date'] = None
        
        results.append(country_data)
    
    return pd.DataFrame(results)

# 监控主要经济体的关键指标
key_indicators = {
    'CPI': '消费者价格指数',
    'GDP': '国内生产总值',
    'URATE': '失业率'
}

major_economies = ['US', 'CN', 'JP', 'DE', 'GB']

monitoring_results = monitor_economic_indicators(major_economies, key_indicators)
print("经济指标监控结果:")
print(monitoring_results)

最佳实践​

  1. 数据验证：始终检查返回的数据质量和完整性
  2. 错误处理：实现适当的错误处理机制
  3. 数据缓存：对于重复使用的数据考虑缓存
  4. 单位注意：注意不同指标的单位和计算方法
  5. 时间对齐：比较不同国家数据时注意时间对齐
# 完整的经济指标分析工作流
def comprehensive_economic_analysis(countries, indicators, years=5):
    """
    综合经济指标分析
    """
    print("开始综合经济指标分析...")
    print(f"分析国家: {', '.join(countries)}")
    print(f"分析指标: {', '.join(indicators.keys())}")
    print(f"分析年限: {years} 年")
    print("-" * 60)
    
    all_data = {}
    
    for country in countries:
        country_data = {}
        print(f"\n分析 {country}:")
        
        for symbol, name in indicators.items():
            try:
                data = obb.economy.indicators(
                    symbol=symbol,
                    country=country,
                    provider="econdb"
                )
                
                df = data.to_dataframe()
                
                if not df.empty:
                    # 获取最近几年的数据
                    recent_data = df.tail(years * 4)  # 假设季度数据
                    
                    country_data[symbol] = {
                        'data': recent_data,
                        'latest_value': recent_data.iloc[-1]['value'],
                        'latest_date': recent_data.iloc[-1]['date'],
                        'trend': 'up' if recent_data['value'].iloc[-1] > recent_data['value'].iloc[0] else 'down'
                    }
                    
                    print(f"  {name}: {country_data[symbol]['latest_value']} (趋势: {country_data[symbol]['trend']})")
                else:
                    print(f"  {name}: 无数据")
                    
            except Exception as e:
                print(f"  {name}: 获取失败 - {e}")
        
        all_data[country] = country_data
    
    return all_data

# 使用示例
analysis_results = comprehensive_economic_analysis(
    countries=['US', 'CN', 'JP'],
    indicators={'CPI': '通胀率', 'GDP': 'GDP', 'URATE': '失业率'},
    years=3
)
最后由 huojiabin 于 更新