第4章 日历模块开发：构建可编程的经济日历

做外汇交易，最怕什么？

怕数据出来那一刻，行情瞬间打穿你的止损。我见过太多人，技术分析画得天花乱坠，结果非农数据一出，直接爆仓。说白了，宏观事件才是外汇市场的真正推手。

这一章，我们就来搞定经济日历模块。它能自动抓取ForexFactory等网站的事件时间与预期值。嗯，这是整个系统的「情报部门」。

4.1 为什么需要自己的经济日历？

你可能会问：「网上免费日历那么多，干嘛自己写？」

我刚开始也这么想。直到有一次，我的策略因为某个小国央行利率决议被打了个措手不及。那个事件在主流日历上根本没显示。你想想看，如果你的交易系统连事件都不知道，怎么提前规避风险？

自己写日历有几个好处：

• 可控性：想抓什么事件就抓什么，不受第三方限制
• 实时性：数据更新频率自己定，毫秒级响应
• 可编程：直接对接交易策略，事件触发自动执行
• 历史数据：积累自己的事件数据库，做回测分析

4.2 整体架构设计

我们先画一张图，看看这个模块长什么样。

这张图把模块分成了5层。从上往下看，数据从网站流进来，经过抓取、解析、存储，最后通过接口喂给交易策略。每一层各司其职，互不干扰。

4.3 数据抓取：跟ForexFactory打交道

ForexFactory的日历页面结构其实挺规整的。我抓了三年多，它的HTML结构基本没大变过。嗯，这里要注意一点——别太频繁请求，不然你的IP会被封。

来看核心代码：

import requests
from bs4 import BeautifulSoup
from datetime import datetime, timedelta
import time

class ForexFactoryCalendar:
    """ForexFactory经济日历抓取器"""
    
    def __init__(self):
        self.base_url = "https://www.forexfactory.com/calendar"
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
                          'AppleWebKit/537.36 (KHTML, like Gecko) '
                          'Chrome/91.0.4472.124 Safari/537.36'
        }
    
    def fetch_events(self, date=None):
        """
        抓取指定日期的事件
        :param date: datetime对象，默认今天
        :return: 事件列表
        """
        if date is None:
            date = datetime.now()
        
        # 构造URL参数
        params = {
            'day': date.strftime('%b%d.%Y').lower()
        }
        
        try:
            response = requests.get(
                self.base_url, 
                params=params,
                headers=self.headers,
                timeout=10
            )
            response.raise_for_status()
            
            # 解析HTML
            soup = BeautifulSoup(response.text, 'html.parser')
            return self._parse_events(soup)
            
        except requests.RequestException as e:
            print(f"抓取失败: {e}")
            return []
    
    def _parse_events(self, soup):
        """解析事件表格"""
        events = []
        table = soup.find('table', class_='calendar__table')
        
        if not table:
            return events
        
        rows = table.find_all('tr', class_='calendar__row')
        
        for row in rows:
            event = self._parse_row(row)
            if event:
                events.append(event)
            
            # 礼貌性延迟
            time.sleep(0.1)
        
        return events
    
    def _parse_row(self, row):
        """解析单行事件"""
        try:
            # 时间
            time_cell = row.find('td', class_='calendar__time')
            if not time_cell:
                return None
            event_time = time_cell.get_text(strip=True)
            
            # 货币
            currency_cell = row.find('td', class_='calendar__currency')
            currency = currency_cell.get_text(strip=True) if currency_cell else ''
            
            # 事件名称
            event_cell = row.find('td', class_='calendar__event')
            event_name = event_cell.get_text(strip=True) if event_cell else ''
            
            # 预期值
            forecast_cell = row.find('td', class_='calendar__forecast')
            forecast = forecast_cell.get_text(strip=True) if forecast_cell else ''
            
            # 前值
            previous_cell = row.find('td', class_='calendar__previous')
            previous = previous_cell.get_text(strip=True) if previous_cell else ''
            
            # 重要性（1-3星）
            impact_cell = row.find('td', class_='calendar__impact')
            impact = len(impact_cell.find_all('span', class_='impact--high')) if impact_cell else 0
            
            return {
                'time': event_time,
                'currency': currency,
                'event': event_name,
                'forecast': forecast,
                'previous': previous,
                'impact': impact,
                'timestamp': datetime.now()
            }
            
        except Exception as e:
            print(f"解析行失败: {e}")
            return None

4.4 数据存储：让事件持久化

抓下来的数据不能丢内存里吧？得存起来。我个人习惯用SQLite，轻量、零配置，单机跑完全够用。

import sqlite3
from datetime import datetime

class CalendarStorage:
    """经济日历数据存储"""
    
    def __init__(self, db_path='calendar.db'):
        self.conn = sqlite3.connect(db_path)
        self._create_tables()
    
    def _create_tables(self):
        """建表"""
        cursor = self.conn.cursor()
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS economic_events (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                event_date DATE NOT NULL,
                event_time TEXT,
                currency TEXT,
                event_name TEXT,
                forecast TEXT,
                previous TEXT,
                actual TEXT,
                impact INTEGER DEFAULT 0,
                source TEXT DEFAULT 'forexfactory',
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                UNIQUE(event_date, event_time, event_name, currency)
            )
        ''')
        
        # 索引加速查询
        cursor.execute('''
            CREATE INDEX IF NOT EXISTS idx_event_date 
            ON economic_events(event_date)
        ''')
        
        cursor.execute('''
            CREATE INDEX IF NOT EXISTS idx_currency 
            ON economic_events(currency)
        ''')
        
        self.conn.commit()
    
    def save_events(self, events, event_date):
        """批量保存事件"""
        cursor = self.conn.cursor()
        
        for event in events:
            try:
                cursor.execute('''
                    INSERT OR IGNORE INTO economic_events 
                    (event_date, event_time, currency, event_name, 
                     forecast, previous, impact)
                    VALUES (?, ?, ?, ?, ?, ?, ?)
                ''', (
                    event_date.strftime('%Y-%m-%d'),
                    event.get('time'),
                    event.get('currency'),
                    event.get('event'),
                    event.get('forecast'),
                    event.get('previous'),
                    event.get('impact')
                ))
            except Exception as e:
                print(f"保存事件失败: {e}")
        
        self.conn.commit()
    
    def get_events(self, date=None, currency=None, min_impact=0):
        """查询事件"""
        cursor = self.conn.cursor()
        
        query = "SELECT * FROM economic_events WHERE 1=1"
        params = []
        
        if date:
            query += " AND event_date = ?"
            params.append(date.strftime('%Y-%m-%d'))
        
        if currency:
            query += " AND currency = ?"
            params.append(currency)
        
        if min_impact > 0:
            query += " AND impact >= ?"
            params.append(min_impact)
        
        query += " ORDER BY event_time"
        
        cursor.execute(query, params)
        return cursor.fetchall()
    
    def close(self):
        self.conn.close()

4.5 定时调度：让日历自动更新

手动跑脚本？太low了。我们要的是自动化。APScheduler这个库，我用下来感觉最顺手。

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
import logging

class CalendarScheduler:
    """日历自动更新调度器"""
    
    def __init__(self):
        self.scheduler = BackgroundScheduler()
        self.calendar = ForexFactoryCalendar()
        self.storage = CalendarStorage()
        
        # 配置日志
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        self.logger = logging.getLogger('CalendarScheduler')
    
    def start(self):
        """启动调度"""
        
        # 每天凌晨2点更新当天数据
        self.scheduler.add_job(
            self._update_today,
            CronTrigger(hour=2, minute=0),
            id='update_today',
            name='更新当天日历'
        )
        
        # 每天凌晨3点更新未来7天数据
        self.scheduler.add_job(
            self._update_week,
            CronTrigger(hour=3, minute=0),
            id='update_week',
            name='更新未来一周日历'
        )
        
        # 每整点检查是否有新数据发布
        self.scheduler.add_job(
            self._check_updates,
            CronTrigger(minute=0),
            id='check_updates',
            name='检查数据更新'
        )
        
        self.scheduler.start()
        self.logger.info("日历调度器已启动")
    
    def _update_today(self):
        """更新今天的数据"""
        self.logger.info("开始更新今天的数据")
        events = self.calendar.fetch_events()
        self.storage.save_events(events, datetime.now())
        self.logger.info(f"今天的数据更新完成，共{len(events)}条")
    
    def _update_week(self):
        """更新未来一周的数据"""
        self.logger.info("开始更新未来一周的数据")
        today = datetime.now()
        
        for i in range(7):
            date = today + timedelta(days=i)
            events = self.calendar.fetch_events(date)
            self.storage.save_events(events, date)
            self.logger.info(f"{date.date()} 更新完成，共{len(events)}条")
    
    def _check_updates(self):
        """检查是否有新数据发布"""
        # 这里可以对比数据库中的最新时间和当前时间
        # 如果有新数据，触发更新
        pass
    
    def stop(self):
        """停止调度"""
        self.scheduler.shutdown()
        self.storage.close()
        self.logger.info("日历调度器已停止")

你想想看，有了这个调度器，你的日历模块就像个永不停歇的情报员。每天自动抓取、自动更新，你只管写策略就行。

4.6 事件驱动：让策略感知风险

数据存好了，怎么用？我设计了一个事件监听器，策略可以订阅特定事件。

import asyncio
from typing import Callable, Dict, List

class EventListener:
    """事件监听器，策略订阅特定事件"""
    
    def __init__(self):
        self.subscribers: Dict[str, List[Callable]] = {}
        self.running = False
    
    def subscribe(self, event_type: str, callback: Callable):
        """
        订阅事件
        :param event_type: 事件类型，如 'NFP', 'FOMC', 'CPI'
        :param callback: 回调函数
        """
        if event_type not in self.subscribers:
            self.subscribers[event_type] = []
        self.subscribers[event_type].append(callback)
        print(f"已订阅事件: {event_type}")
    
    def unsubscribe(self, event_type: str, callback: Callable):
        """取消订阅"""
        if event_type in self.subscribers:
            self.subscribers[event_type].remove(callback)
    
    async def notify(self, event: dict):
        """
        通知所有订阅者
        :param event: 事件数据
        """
        event_name = event.get('event', '')
        
        # 检查是否有匹配的订阅
        for event_type, callbacks in self.subscribers.items():
            if event_type in event_name.upper():
                for callback in callbacks:
                    try:
                        await callback(event)
                    except Exception as e:
                        print(f"回调执行失败: {e}")
    
    async def start_monitoring(self, check_interval=60):
        """开始监控新事件"""
        self.running = True
        storage = CalendarStorage()
        
        while self.running:
            # 获取未来1小时内的事件
            now = datetime.now()
            one_hour_later = now + timedelta(hours=1)
            
            events = storage.get_events(
                date=now,
                min_impact=2  # 只关注高重要性事件
            )
            
            for event in events:
                await self.notify({
                    'event': event[4],  # event_name
                    'currency': event[3],
                    'time': event[2],
                    'forecast': event[5],
                    'previous': event[6],
                    'impact': event[8]
                })
            
            await asyncio.sleep(check_interval)
        
        storage.close()
    
    def stop(self):
        self.running = False

4.7 完整集成示例

最后，把上面所有模块串起来：

def main():
    """启动日历模块"""
    
    # 1. 初始化调度器
    scheduler = CalendarScheduler()
    
    # 2. 启动定时任务
    scheduler.start()
    
    # 3. 初始化事件监听
    listener = EventListener()
    
    # 4. 策略订阅事件
    async def on_nfp(event):
        print(f"非农数据来了: {event}")
        # 这里写你的交易逻辑
    
    listener.subscribe('NFP', on_nfp)
    
    # 5. 启动监控
    asyncio.run(listener.start_monitoring())
    
    try:
        # 保持程序运行
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        scheduler.stop()
        listener.stop()
        print("日历模块已关闭")

if __name__ == '__main__':
    main()

嗯，到这里，日历模块的核心就讲完了。说白了，就是抓数据、存数据、用数据这三步。但每一步都有坑，我踩过的那些坑都写在代码注释里了。你照着写，应该能少走不少弯路。

记住一点：经济日历不是摆设，它是你交易系统的眼睛。眼睛亮了，策略才能看得清方向。

---

公众号:蓝海资料掘金营，微信deep3321