跳到主要内容

市场日历

市场日历是任何分析师日常工作的重要组成部分。经济事件和公司行为提供了近期交易机会（或退出的理由），因为预期与现实相遇。OpenBB 平台有多种日历。

  • 经济日历
  • 财报日历
  • 股息日历
  • 拆股日历
  • IPO/SPO 日历

让我们看看一些用例。

备注

本页面的示例假设已安装 OpenBB 平台，Python 环境处于活动状态，并且 Python 接口已导入到活动会话中。

from openbb import obb
import pandas as pd

经济日历​

Details

经济日历汇总了全球央行和宏观经济发布信息，位于 obb.economy 模块内。

# 获取本周的经济事件
economic_calendar = obb.economy.calendar(provider="fmp")
df_economic = economic_calendar.to_dataframe()

print("本周经济事件:")
print(df_economic.head(10))

# 筛选重要事件
if 'importance' in df_economic.columns:
    important_events = df_economic[df_economic['importance'] == 'High']
    print(f"\n重要经济事件 ({len(important_events)} 个):")
    print(important_events[['date', 'event', 'country']].head())

按国家筛选经济事件​

def get_economic_events_by_country(country_code, days_ahead=7):
    """
    获取特定国家的经济事件
    """
    from datetime import datetime, timedelta
    
    try:
        # 计算日期范围
        start_date = datetime.now()
        end_date = start_date + timedelta(days=days_ahead)
        
        # 获取经济日历
        calendar = obb.economy.calendar(
            start_date=start_date.strftime("%Y-%m-%d"),
            end_date=end_date.strftime("%Y-%m-%d"),
            provider="fmp"
        )
        
        df = calendar.to_dataframe()
        
        if df.empty:
            print(f"未找到 {country_code} 的经济事件")
            return None
        
        # 按国家筛选
        if 'country' in df.columns:
            country_events = df[df['country'].str.contains(country_code, case=False, na=False)]
            
            if not country_events.empty:
                print(f"{country_code} 未来 {days_ahead} 天的经济事件:")
                for _, event in country_events.iterrows():
                    print(f"  {event['date']}: {event['event']}")
                
                return country_events
        
        return None
        
    except Exception as e:
        print(f"获取 {country_code} 经济事件失败: {e}")
        return None

# 获取美国经济事件
us_events = get_economic_events_by_country("US", days_ahead=14)

# 获取中国经济事件
china_events = get_economic_events_by_country("CN", days_ahead=14)

财报日历​

# 获取财报日历
earnings_calendar = obb.equity.calendar.earnings(provider="fmp")
df_earnings = earnings_calendar.to_dataframe()

print("即将发布财报的公司:")
print(df_earnings.head(10))

# 按日期分组
if 'date' in df_earnings.columns:
    earnings_by_date = df_earnings.groupby('date').size()
    print(f"\n各日期财报发布数量:")
    print(earnings_by_date.head())

重点关注的财报​

def track_earnings_for_symbols(symbols, days_ahead=30):
    """
    跟踪特定股票的财报发布日期
    """
    from datetime import datetime, timedelta
    
    try:
        # 获取财报日历
        earnings = obb.equity.calendar.earnings(provider="fmp")
        df = earnings.to_dataframe()
        
        if df.empty:
            print("未获取到财报日历数据")
            return None
        
        # 筛选关注的股票
        if 'symbol' in df.columns:
            tracked_earnings = df[df['symbol'].isin(symbols)]
            
            if not tracked_earnings.empty:
                print("关注股票的财报日程:")
                for _, earning in tracked_earnings.iterrows():
                    print(f"  {earning['symbol']}: {earning['date']}")
                    if 'eps_estimate' in earning:
                        print(f"    预期EPS: {earning['eps_estimate']}")
                
                return tracked_earnings
            else:
                print("未找到关注股票的财报信息")
        
        return None
        
    except Exception as e:
        print(f"跟踪财报失败: {e}")
        return None

# 跟踪科技股财报
tech_stocks = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'META']
tech_earnings = track_earnings_for_symbols(tech_stocks)

股息日历​

# 获取股息日历
dividend_calendar = obb.equity.calendar.dividend(provider="fmp")
df_dividends = dividend_calendar.to_dataframe()

print("即将发放股息的公司:")
print(df_dividends.head(10))

# 计算股息收益率
def calculate_dividend_yield(symbol, dividend_amount):
    """
    计算股息收益率
    """
    try:
        # 获取当前股价
        quote = obb.equity.price.quote(symbol=symbol, provider="yfinance")
        current_price = quote.to_dataframe()['last_price'].iloc[0]
        
        # 计算年化股息收益率（假设季度分红）
        annual_dividend = dividend_amount * 4
        dividend_yield = (annual_dividend / current_price) * 100
        
        return dividend_yield
        
    except Exception as e:
        print(f"计算 {symbol} 股息收益率失败: {e}")
        return None

# 分析高股息股票
if not df_dividends.empty and 'dividend' in df_dividends.columns:
    high_dividend_stocks = df_dividends[df_dividends['dividend'] > 1.0]  # 股息超过1美元
    
    print("\n高股息股票分析:")
    for _, stock in high_dividend_stocks.head(5).iterrows():
        symbol = stock['symbol']
        dividend = stock['dividend']
        yield_rate = calculate_dividend_yield(symbol, dividend)
        
        if yield_rate:
            print(f"  {symbol}: 股息 ${dividend:.2f}, 收益率 {yield_rate:.2f}%")

IPO 日历​

# 获取 IPO 日历
try:
    ipo_calendar = obb.equity.calendar.ipo(provider="fmp")
    df_ipo = ipo_calendar.to_dataframe()
    
    print("即将上市的公司:")
    print(df_ipo.head(10))
    
    # 按价格区间分析
    if 'price_range_low' in df_ipo.columns and 'price_range_high' in df_ipo.columns:
        df_ipo['price_range_mid'] = (df_ipo['price_range_low'] + df_ipo['price_range_high']) / 2
        
        # 大型IPO（估值超过10亿）
        large_ipos = df_ipo[df_ipo['price_range_mid'] * df_ipo.get('shares_offered', 1) > 1000000000]
        
        if not large_ipos.empty:
            print("\n大型IPO项目:")
            for _, ipo in large_ipos.iterrows():
                print(f"  {ipo['symbol']}: {ipo['company_name']}")
                print(f"    价格区间: ${ipo['price_range_low']:.2f} - ${ipo['price_range_high']:.2f}")

except Exception as e:
    print(f"获取IPO日历失败: {e}")

拆股日历​

# 获取拆股日历
try:
    splits_calendar = obb.equity.calendar.splits(provider="fmp")
    df_splits = splits_calendar.to_dataframe()
    
    print("即将拆股的公司:")
    print(df_splits.head(10))
    
    # 分析拆股比例
    if 'split_ratio' in df_splits.columns:
        # 高比例拆股（1:5以上）
        high_ratio_splits = df_splits[df_splits['split_ratio'].str.contains('1:5|1:10|1:20', na=False)]
        
        if not high_ratio_splits.empty:
            print("\n高比例拆股:")
            for _, split in high_ratio_splits.iterrows():
                print(f"  {split['symbol']}: {split['split_ratio']} ({split['date']})")

except Exception as e:
    print(f"获取拆股日历失败: {e}")

综合市场日历分析​

def comprehensive_market_calendar(days_ahead=7):
    """
    综合市场日历分析
    """
    from datetime import datetime, timedelta
    
    print(f"未来 {days_ahead} 天市场日历综合分析")
    print("=" * 60)
    
    end_date = datetime.now() + timedelta(days=days_ahead)
    
    # 1. 经济事件
    print("1. 重要经济事件:")
    try:
        economic = obb.economy.calendar(
            end_date=end_date.strftime("%Y-%m-%d"),
            provider="fmp"
        )
        df_econ = economic.to_dataframe()
        
        if not df_econ.empty:
            # 筛选重要事件
            if 'importance' in df_econ.columns:
                important = df_econ[df_econ['importance'] == 'High']
                print(f"   发现 {len(important)} 个重要经济事件")
                for _, event in important.head(5).iterrows():
                    print(f"   • {event['date']}: {event['event']} ({event['country']})")
            else:
                print(f"   发现 {len(df_econ)} 个经济事件")
        else:
            print("   未找到经济事件")
    except Exception as e:
        print(f"   获取经济事件失败: {e}")
    
    # 2. 财报发布
    print("\n2. 重要财报发布:")
    try:
        earnings = obb.equity.calendar.earnings(provider="fmp")
        df_earn = earnings.to_dataframe()
        
        if not df_earn.empty:
            # 筛选大市值公司
            large_cap_symbols = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'TSLA', 'META', 'NVDA']
            if 'symbol' in df_earn.columns:
                large_cap_earnings = df_earn[df_earn['symbol'].isin(large_cap_symbols)]
                
                if not large_cap_earnings.empty:
                    print(f"   发现 {len(large_cap_earnings)} 个大市值公司财报")
                    for _, earning in large_cap_earnings.head(5).iterrows():
                        print(f"   • {earning['symbol']}: {earning['date']}")
                else:
                    print(f"   发现 {len(df_earn)} 个公司财报")
        else:
            print("   未找到财报信息")
    except Exception as e:
        print(f"   获取财报信息失败: {e}")
    
    # 3. 股息发放
    print("\n3. 股息发放:")
    try:
        dividends = obb.equity.calendar.dividend(provider="fmp")
        df_div = dividends.to_dataframe()
        
        if not df_div.empty:
            # 筛选高股息
            if 'dividend' in df_div.columns:
                high_div = df_div[df_div['dividend'] > 0.5]  # 股息超过0.5美元
                print(f"   发现 {len(high_div)} 个高股息股票")
                for _, div in high_div.head(5).iterrows():
                    print(f"   • {div['symbol']}: ${div['dividend']:.2f} ({div['date']})")
            else:
                print(f"   发现 {len(df_div)} 个股息发放")
        else:
            print("   未找到股息信息")
    except Exception as e:
        print(f"   获取股息信息失败: {e}")
    
    print(f"\n市场日历分析完成!")

# 运行综合分析
comprehensive_market_calendar(days_ahead=14)

日历提醒系统​

def create_market_alerts(watch_symbols=None, alert_days=3):
    """
    创建市场事件提醒系统
    """
    from datetime import datetime, timedelta
    
    if watch_symbols is None:
        watch_symbols = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'TSLA']
    
    alerts = []
    alert_date = datetime.now() + timedelta(days=alert_days)
    
    print(f"设置 {alert_days} 天内的市场提醒...")
    
    # 财报提醒
    try:
        earnings = obb.equity.calendar.earnings(provider="fmp")
        df_earnings = earnings.to_dataframe()
        
        if not df_earnings.empty and 'symbol' in df_earnings.columns:
            watched_earnings = df_earnings[df_earnings['symbol'].isin(watch_symbols)]
            
            for _, earning in watched_earnings.iterrows():
                alert = {
                    'type': '财报发布',
                    'symbol': earning['symbol'],
                    'date': earning['date'],
                    'details': f"预期EPS: {earning.get('eps_estimate', 'N/A')}"
                }
                alerts.append(alert)
    
    except Exception as e:
        print(f"设置财报提醒失败: {e}")
    
    # 股息提醒
    try:
        dividends = obb.equity.calendar.dividend(provider="fmp")
        df_dividends = dividends.to_dataframe()
        
        if not df_dividends.empty and 'symbol' in df_dividends.columns:
            watched_dividends = df_dividends[df_dividends['symbol'].isin(watch_symbols)]
            
            for _, dividend in watched_dividends.iterrows():
                alert = {
                    'type': '股息发放',
                    'symbol': dividend['symbol'],
                    'date': dividend['date'],
                    'details': f"股息: ${dividend.get('dividend', 'N/A'):.2f}"
                }
                alerts.append(alert)
    
    except Exception as e:
        print(f"设置股息提醒失败: {e}")
    
    # 显示提醒
    if alerts:
        print(f"\n发现 {len(alerts)} 个提醒事件:")
        for alert in sorted(alerts, key=lambda x: x['date']):
            print(f"  {alert['date']}: {alert['symbol']} - {alert['type']}")
            print(f"    {alert['details']}")
    else:
        print("未发现需要提醒的事件")
    
    return alerts

# 创建提醒
my_watchlist = ['AAPL', 'MSFT', 'GOOGL', 'NVDA', 'TSLA']
alerts = create_market_alerts(my_watchlist, alert_days=7)

最佳实践​

  1. 时区处理：注意不同市场的时区差异
  2. 数据更新：市场日历数据会频繁更新，及时获取最新信息
  3. 重要性筛选：关注高重要性的经济事件
  4. 组合管理：根据持仓情况关注相关的市场事件
  5. 提前准备：提前了解重要事件，制定交易策略
# 完整的市场日历监控系统
class MarketCalendarMonitor:
    def __init__(self, watchlist=None):
        self.watchlist = watchlist or []
        self.alerts = []
    
    def update_economic_calendar(self, days_ahead=7):
        """更新经济日历"""
        try:
            calendar = obb.economy.calendar(provider="fmp")
            self.economic_events = calendar.to_dataframe()
            print(f"✓ 经济日历更新完成，共 {len(self.economic_events)} 个事件")
        except Exception as e:
            print(f"✗ 经济日历更新失败: {e}")
    
    def update_earnings_calendar(self):
        """更新财报日历"""
        try:
            earnings = obb.equity.calendar.earnings(provider="fmp")
            self.earnings_events = earnings.to_dataframe()
            print(f"✓ 财报日历更新完成，共 {len(self.earnings_events)} 个事件")
        except Exception as e:
            print(f"✗ 财报日历更新失败: {e}")
    
    def check_alerts(self):
        """检查提醒事件"""
        alerts = []
        
        # 检查关注股票的财报
        if hasattr(self, 'earnings_events') and not self.earnings_events.empty:
            if 'symbol' in self.earnings_events.columns:
                watched_earnings = self.earnings_events[
                    self.earnings_events['symbol'].isin(self.watchlist)
                ]
                
                for _, earning in watched_earnings.iterrows():
                    alerts.append({
                        'type': 'earnings',
                        'symbol': earning['symbol'],
                        'date': earning['date'],
                        'message': f"{earning['symbol']} 财报发布"
                    })
        
        self.alerts = alerts
        return alerts
    
    def daily_summary(self):
        """每日市场摘要"""
        print("每日市场日历摘要")
        print("=" * 40)
        
        self.update_economic_calendar()
        self.update_earnings_calendar()
        
        alerts = self.check_alerts()
        
        if alerts:
            print(f"\n发现 {len(alerts)} 个关注事件:")
            for alert in alerts:
                print(f"  • {alert['date']}: {alert['message']}")
        else:
            print("\n今日无特别关注事件")

# 使用示例
monitor = MarketCalendarMonitor(['AAPL', 'MSFT', 'GOOGL', 'TSLA'])
monitor.daily_summary()
最后由 huojiabin 于 更新