feat(calendar): add script to generate calendar
This commit is contained in:
		
							
								
								
									
										1
									
								
								.gitignore
									
									
									
									
										vendored
									
									
								
							
							
						
						
									
										1
									
								
								.gitignore
									
									
									
									
										vendored
									
									
								
							@@ -1,2 +1,3 @@
 | 
				
			|||||||
output/
 | 
					output/
 | 
				
			||||||
 | 
					public/
 | 
				
			||||||
__pycache__/
 | 
					__pycache__/
 | 
				
			||||||
							
								
								
									
										135
									
								
								generate_calendar.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										135
									
								
								generate_calendar.py
									
									
									
									
									
										Normal file
									
								
							@@ -0,0 +1,135 @@
 | 
				
			|||||||
 | 
					"""根据交易配置生成ICS日历文件。"""
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					import json
 | 
				
			||||||
 | 
					import logging
 | 
				
			||||||
 | 
					import os
 | 
				
			||||||
 | 
					from datetime import datetime
 | 
				
			||||||
 | 
					from icalendar import Calendar, Event
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					# 配置日志
 | 
				
			||||||
 | 
					logging.basicConfig(
 | 
				
			||||||
 | 
					    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
 | 
				
			||||||
 | 
					)
 | 
				
			||||||
 | 
					logger = logging.getLogger(__name__)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					def load_config(config_path="trading_config.json"):
 | 
				
			||||||
 | 
					    """加载交易配置文件。
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    Args:
 | 
				
			||||||
 | 
					        config_path (str): 配置文件路径
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    Returns:
 | 
				
			||||||
 | 
					        dict: 解析后的配置字典
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    Raises:
 | 
				
			||||||
 | 
					        FileNotFoundError: 当文件不存在时
 | 
				
			||||||
 | 
					        json.JSONDecodeError: 当JSON解析失败时
 | 
				
			||||||
 | 
					    """
 | 
				
			||||||
 | 
					    try:
 | 
				
			||||||
 | 
					        with open(config_path, "r", encoding="utf-8") as file:
 | 
				
			||||||
 | 
					            return json.load(file)
 | 
				
			||||||
 | 
					    except FileNotFoundError:
 | 
				
			||||||
 | 
					        logger.error("配置文件不存在: %s", config_path)
 | 
				
			||||||
 | 
					        raise
 | 
				
			||||||
 | 
					    except json.JSONDecodeError as ex:
 | 
				
			||||||
 | 
					        logger.error("配置文件不是有效的JSON: %s", ex)
 | 
				
			||||||
 | 
					        raise
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					def create_event(symbol, comment, date_obj, description):
 | 
				
			||||||
 | 
					    """创建单个日历事件。
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    Args:
 | 
				
			||||||
 | 
					        symbol (str): 交易对符号
 | 
				
			||||||
 | 
					        comment (str): 交易备注
 | 
				
			||||||
 | 
					        date_obj (date): 交易日期
 | 
				
			||||||
 | 
					        description (str): 事件描述
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    Returns:
 | 
				
			||||||
 | 
					        Event: 创建好的事件对象
 | 
				
			||||||
 | 
					    """
 | 
				
			||||||
 | 
					    event = Event()
 | 
				
			||||||
 | 
					    event.add("uid", f"{symbol.lower()}-{date_obj.strftime('%Y%m%d')}@trade")
 | 
				
			||||||
 | 
					    event.add("dtstamp", datetime.now())
 | 
				
			||||||
 | 
					    event.add("dtstart", date_obj)
 | 
				
			||||||
 | 
					    event.add("summary", f"{symbol} 交易 ({comment})")
 | 
				
			||||||
 | 
					    event.add("description", description)
 | 
				
			||||||
 | 
					    return event
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					def generate_ics(config):
 | 
				
			||||||
 | 
					    """生成ICS日历内容。
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    Args:
 | 
				
			||||||
 | 
					        config (dict): 交易配置
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    Returns:
 | 
				
			||||||
 | 
					        Calendar: 生成的日历对象
 | 
				
			||||||
 | 
					    """
 | 
				
			||||||
 | 
					    cal = Calendar()
 | 
				
			||||||
 | 
					    cal.add("prodid", "-//Trading Calendar//EN")
 | 
				
			||||||
 | 
					    cal.add("version", "2.0")
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    symbol_mapping = config.get("symbol_mapping", {})
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    for trade in config.get("trades", []):
 | 
				
			||||||
 | 
					        symbol = symbol_mapping.get(trade["symbol"], trade["symbol"])
 | 
				
			||||||
 | 
					        order_type = trade["order_type"]
 | 
				
			||||||
 | 
					        side = trade["side"]
 | 
				
			||||||
 | 
					        comment = trade["comment"]
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					        # 构建描述
 | 
				
			||||||
 | 
					        amount = trade["params"].get("quoteOrderQty")
 | 
				
			||||||
 | 
					        quantity = trade["params"].get("quantity")
 | 
				
			||||||
 | 
					        description = f"交易类型: {order_type}\n方向: {side}"
 | 
				
			||||||
 | 
					        if quantity:
 | 
				
			||||||
 | 
					            description = f"{description}\n数量: {quantity}"
 | 
				
			||||||
 | 
					        if amount:
 | 
				
			||||||
 | 
					            description = f"{description}\n金额: {amount}"
 | 
				
			||||||
 | 
					        description = f"{description}\n备注: {comment}"
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					        if trade["execute_dates"] == ["*"]:
 | 
				
			||||||
 | 
					            event = create_event(symbol, comment, datetime.now().date(), description)
 | 
				
			||||||
 | 
					            event.add("rrule", {"freq": "daily"})
 | 
				
			||||||
 | 
					            cal.add_component(event)
 | 
				
			||||||
 | 
					            logger.info("已创建项目:%s 交易 (%s) 每日", symbol, comment)
 | 
				
			||||||
 | 
					        else:
 | 
				
			||||||
 | 
					            for date_str in trade["execute_dates"]:
 | 
				
			||||||
 | 
					                try:
 | 
				
			||||||
 | 
					                    date_obj = datetime.strptime(date_str, "%Y-%m-%d").date()
 | 
				
			||||||
 | 
					                    event = create_event(symbol, comment, date_obj, description)
 | 
				
			||||||
 | 
					                    cal.add_component(event)
 | 
				
			||||||
 | 
					                    logger.info("已创建项目:%s 交易 (%s) %s", symbol, comment, date_str)
 | 
				
			||||||
 | 
					                except ValueError as ex:
 | 
				
			||||||
 | 
					                    logger.warning("跳过无效日期 %s: %s", date_str, ex)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    return cal
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					def save_ics(calendar, output_path="public/trading.ics"):
 | 
				
			||||||
 | 
					    """保存ICS文件。
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					    Args:
 | 
				
			||||||
 | 
					        calendar (Calendar): 日历对象
 | 
				
			||||||
 | 
					        output_path (str): 输出路径
 | 
				
			||||||
 | 
					    """
 | 
				
			||||||
 | 
					    os.makedirs(os.path.dirname(output_path), exist_ok=True)
 | 
				
			||||||
 | 
					    with open(output_path, "wb") as file:
 | 
				
			||||||
 | 
					        file.write(calendar.to_ical())
 | 
				
			||||||
 | 
					    logger.info("日历文件已生成: %s", output_path)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					def main():
 | 
				
			||||||
 | 
					    """主执行函数。"""
 | 
				
			||||||
 | 
					    try:
 | 
				
			||||||
 | 
					        config = load_config()
 | 
				
			||||||
 | 
					        calendar = generate_ics(config)
 | 
				
			||||||
 | 
					        save_ics(calendar)
 | 
				
			||||||
 | 
					    except Exception as ex:  # pylint: disable=broad-except
 | 
				
			||||||
 | 
					        logger.error("生成失败: %s", ex, exc_info=True)
 | 
				
			||||||
 | 
					        raise
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					if __name__ == "__main__":
 | 
				
			||||||
 | 
					    main()
 | 
				
			||||||
		Reference in New Issue
	
	Block a user