158 lines
5.0 KiB
Python
158 lines
5.0 KiB
Python
"""根据交易配置生成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(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():
|
||
"""主执行函数，遍历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()
|