feat(tool): CSV merge trades for the same pair on the same day

tool_generate_calendar.py

@@ -0,0 +1,159 @@
+"""根据交易配置生成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="config/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(summary: str, date_obj, description):
+    """创建单个日历事件。
+
+    Args:
+        summary (str): 事件标题
+        date_obj (date): 交易日期
+        description (str): 事件描述
+
+    Returns:
+        Event: 创建好的事件对象
+    """
+    event = Event()
+    event.add(
+        "uid",
+        f"{summary.lower().replace('/', '-')}-{date_obj.strftime('%Y%m%d')}@trade",
+    )
+    event.add("dtstamp", datetime.now())
+    event.add("dtstart", date_obj)
+    event.add("summary", summary)
+    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")
+
+    for trade in config.get("trades", []):
+        symbol = trade["symbol"]
+        order_type = trade["order_type"]
+        side = trade["side"]
+        comment = trade["comment"]
+
+        # 构建描述
+        amount = trade["params"].get("quoteOrderQty")
+        quantity = trade["params"].get("quantity")
+
+        if trade["execute_dates"] == ["*"]:
+            summary = f"{symbol}/{order_type}/{side}"
+            if amount:
+                summary = f"{summary}/{amount}{symbol[-4:]}"
+            if quantity:
+                summary = f"{summary}/{quantity}{symbol[:-4]}"
+            event = create_event(summary, datetime.now().date(), comment)
+            event.add("rrule", {"freq": "daily"})
+            cal.add_component(event)
+            logger.info("已创建项目：每日 %s", summary)
+        else:
+            for date_str in trade["execute_dates"]:
+                try:
+                    date_obj = datetime.strptime(date_str, "%Y-%m-%d").date()
+                    summary = f"{symbol}/{order_type}/{side}"
+                    if amount:
+                        summary = f"{summary}/{amount}{symbol[-4:]}"
+                    if quantity:
+                        summary = f"{summary}/{quantity}{symbol[:-4]}"
+                    event = create_event(summary, date_obj, comment)
+                    cal.add_component(event)
+                    logger.info("已创建项目：%s %s", date_str, summary)
+                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():
+    """主执行函数，遍历config目录下的所有json文件并生成对应的ics文件。"""
+
+    try:
+        # 遍历config目录下的所有json文件
+        for filename in os.listdir("config"):
+            if filename.endswith(".json"):
+                config_path = os.path.join("config", filename)
+                try:
+                    # 加载配置
+                    config = load_config(config_path)
+
+                    # 生成ics文件名（保留原文件名，只改扩展名）
+                    ics_filename = os.path.splitext(filename)[0] + ".ics"
+                    ics_path = os.path.join("public", ics_filename)
+
+                    # 生成并保存ics文件
+                    calendar = generate_ics(config)
+                    save_ics(calendar, ics_path)
+
+                    logger.info("成功生成: %s → %s", config_path, ics_path)
+                except Exception as ex:
+                    logger.error("处理文件 %s 失败: %s", config_path, ex, exc_info=True)
+                    continue  # 继续处理下一个文件
+
+        logger.info("所有文件处理完成")
+    except Exception as ex:
+        logger.error("程序执行失败: %s", ex, exc_info=True)
+        raise
+
+
+if __name__ == "__main__":
+    main()