diff --git a/scripts/daily_scheduler.py b/scripts/daily_scheduler.py index 4b054a6..658219e 100644 --- a/scripts/daily_scheduler.py +++ b/scripts/daily_scheduler.py @@ -26,6 +26,7 @@ project_root = Path(__file__).parent.parent sys.path.insert(0, str(project_root)) from dotenv import load_dotenv + load_dotenv(project_root / ".env") from loguru import logger @@ -64,13 +65,10 @@ def is_trading_day(date_str: str = None) -> bool: pro = ts.pro_api(token) df = pro.trade_cal( - exchange='SSE', - start_date=date_str, - end_date=date_str, - is_open='1' + exchange="SSE", start_date=date_str, end_date=date_str, is_open="1" ) - is_open = len(df) > 0 and df.iloc[0]['is_open'] == 1 + is_open = len(df) > 0 and df.iloc[0]["is_open"] == 1 logger.info(f"日期 {date_str} 是否为交易日: {is_open}") return is_open @@ -97,8 +95,10 @@ def run_strategy(config_path: str = "config/strategies/rotation.yaml") -> dict: cmd = [ sys.executable, str(project_root / "scripts" / "run_rotation.py"), - "--config", config_path, - "--save-path", f"results/report_{datetime.now().strftime('%Y%m%d')}" + "--config", + config_path, + "--save-path", + f"results/report_{datetime.now().strftime('%Y%m%d')}", ] logger.info(f"执行命令: {' '.join(cmd)}") @@ -109,7 +109,7 @@ def run_strategy(config_path: str = "config/strategies/rotation.yaml") -> dict: capture_output=True, text=True, cwd=project_root, - timeout=300 # 5分钟超时 + timeout=300, # 5分钟超时 ) if result.returncode != 0: @@ -120,7 +120,7 @@ def run_strategy(config_path: str = "config/strategies/rotation.yaml") -> dict: logger.debug(result.stdout) # 查找生成的报告文件 - report_date = datetime.now().strftime('%Y%m%d') + report_date = datetime.now().strftime("%Y%m%d") chart_path = project_root / "results" / f"report_{report_date}_chart.png" # 如果找不到带日期的,尝试默认路径 @@ -161,14 +161,14 @@ def send_report_to_dingtalk(chart_path: str, summary_text: str = "") -> bool: logger.error("钉钉未配置,无法发送") return False - today_str = datetime.now().strftime('%Y-%m-%d') + today_str = datetime.now().strftime("%Y-%m-%d") # 发送图文消息 success = bot.send_image_via_oss( image_path=chart_path, title=f"ETF轮动策略日报 ({today_str})", text=summary_text or f"今日调仓信号已生成", - expire_days=7 + expire_days=7, ) if success: @@ -183,7 +183,9 @@ def send_report_to_dingtalk(chart_path: str, summary_text: str = "") -> bool: return False -def setup_schedule(target_time: str = "15:30", config_path: str = "config/strategies/rotation.yaml"): +def setup_schedule( + target_time: str = "15:30", config_path: str = "config/strategies/rotation.yaml" +): """ 设置定时任务 @@ -209,7 +211,7 @@ def daily_task(config_path: str = "config/strategies/rotation.yaml"): Args: config_path: 配置文件路径 """ - today_str = datetime.now().strftime('%Y-%m-%d') + today_str = datetime.now().strftime("%Y-%m-%d") logger.info(f"=" * 60) logger.info(f"开始执行每日任务: {today_str}") logger.info(f"=" * 60) @@ -231,8 +233,7 @@ def daily_task(config_path: str = "config/strategies/rotation.yaml"): # 3. 发送报告 if result.get("chart_path"): send_report_to_dingtalk( - chart_path=result["chart_path"], - summary_text="今日ETF轮动策略调仓信号" + chart_path=result["chart_path"], summary_text="今日ETF轮动策略调仓信号" ) else: logger.warning("未找到报告图表") @@ -256,7 +257,7 @@ def main(): parser.add_argument( "--time", type=str, - default="15:30", + default="09:00", help="执行时间 (HH:MM),默认15:30", ) parser.add_argument(