feat(scripts): 迁移轮动策略定时调度器
新增文件: - scripts/daily_scheduler.py: 定时调度器,支持交易日判断、回测执行、OSS上传、钉钉推送 - scripts/run_rotation.py: 回测入口脚本,支持Flask API和本地数据源切换 - config/settings.py: 配置管理模块,支持钉钉多群配置 功能: 1. 每天15:30自动检查交易日 2. 交易日执行策略回测生成报告 3. 上传报告图片到OSS 4. 发送图片链接到钉钉群 修复: - 添加oss2库SyntaxWarning过滤(Python 3.12兼容) - 钉钉消息精简为标题+图片格式
This commit is contained in:
93
config/settings.py
Normal file
93
config/settings.py
Normal file
@@ -0,0 +1,93 @@
|
||||
"""
|
||||
配置管理模块
|
||||
|
||||
从环境变量读取配置信息
|
||||
"""
|
||||
|
||||
import os
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
# 默认基准
|
||||
DEFAULT_BENCHMARK_CODE = "000300.SH"
|
||||
DEFAULT_BENCHMARK_NAME = "沪深300指数"
|
||||
|
||||
# 默认代码名称映射
|
||||
DEFAULT_CODE_NAME_MAP = {
|
||||
"000300.SH": "沪深300",
|
||||
"000905.SH": "中证500",
|
||||
"000852.SH": "中证1000",
|
||||
}
|
||||
|
||||
|
||||
def get_dingtalk_config() -> Dict[str, str]:
|
||||
"""
|
||||
获取钉钉机器人配置(第一个群)
|
||||
|
||||
Returns:
|
||||
dict: 包含 webhook 和 secret
|
||||
"""
|
||||
webhook = os.getenv("DINGTALK_WEBHOOK", "")
|
||||
secret = os.getenv("DINGTALK_SECRET", "")
|
||||
|
||||
return {
|
||||
"webhook": webhook,
|
||||
"secret": secret,
|
||||
}
|
||||
|
||||
|
||||
def get_all_dingtalk_configs() -> List[Dict[str, str]]:
|
||||
"""
|
||||
获取所有钉钉机器人配置(支持多群)
|
||||
|
||||
环境变量格式:
|
||||
DINGTALK_WEBHOOK_1=xxx
|
||||
DINGTALK_SECRET_1=xxx
|
||||
DINGTALK_WEBHOOK_2=xxx
|
||||
DINGTALK_SECRET_2=xxx
|
||||
...
|
||||
|
||||
如果没有编号配置,则使用 DINGTALK_WEBHOOK 和 DINGTALK_SECRET
|
||||
|
||||
Returns:
|
||||
list: 配置列表,每项包含 webhook 和 secret
|
||||
"""
|
||||
configs = []
|
||||
|
||||
# 查找编号配置
|
||||
i = 1
|
||||
while True:
|
||||
webhook = os.getenv(f"DINGTALK_WEBHOOK_{i}", "")
|
||||
secret = os.getenv(f"DINGTALK_SECRET_{i}", "")
|
||||
|
||||
if not webhook:
|
||||
break
|
||||
|
||||
configs.append({
|
||||
"webhook": webhook,
|
||||
"secret": secret,
|
||||
})
|
||||
i += 1
|
||||
|
||||
# 如果没有编号配置,使用默认配置
|
||||
if not configs:
|
||||
default_config = get_dingtalk_config()
|
||||
if default_config["webhook"]:
|
||||
configs.append(default_config)
|
||||
|
||||
return configs
|
||||
|
||||
|
||||
def get_db_config() -> Dict[str, str]:
|
||||
"""
|
||||
获取数据库配置
|
||||
|
||||
Returns:
|
||||
dict: 数据库配置
|
||||
"""
|
||||
return {
|
||||
"host": os.getenv("DB_HOST", "localhost"),
|
||||
"port": os.getenv("DB_PORT", "5432"),
|
||||
"user": os.getenv("DB_USER", ""),
|
||||
"password": os.getenv("DB_PASSWORD", ""),
|
||||
"database": os.getenv("DB_NAME", ""),
|
||||
}
|
||||
Reference in New Issue
Block a user