From e7ca36d968e798f2075b2c531b8cf19e9fa08e39 Mon Sep 17 00:00:00 2001
From: Zichao Lin <earthjasonlin@163.com>
Date: Thu, 17 Jul 2025 00:15:34 +0800
Subject: [PATCH] feat(calendar): add script to generate calendar

---
 .gitignore           |   1 +
 generate_calendar.py | 135 +++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 136 insertions(+)
 create mode 100644 generate_calendar.py

diff --git a/.gitignore b/.gitignore
index 1f93644..fbb00e1 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,2 +1,3 @@
 output/
+public/
 __pycache__/
\ No newline at end of file
diff --git a/generate_calendar.py b/generate_calendar.py
new file mode 100644
index 0000000..97007f7
--- /dev/null
+++ b/generate_calendar.py
@@ -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()