fix(scheduler): 修复格式和代码风格问题,调整默认执行时间

- 规范了代码中的字符串引号统一使用双引号
- 调整了部分代码的参数换行和缩进格式
- 在load_dotenv调用中添加了缺失的空行
- 优化了日志打印字符串内引号使用一致性
- 修改主函数默认执行时间由15:30调整为09:00
- 修正字典取值和列表拼接中的引号使用,避免潜在错误
- 规范函数定义参数列表的格式,提高代码可读性
This commit is contained in:
2026-03-19 21:57:24 +08:00
parent 8d24fb91eb
commit 76feec6824

View File

@@ -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(