chore(config): 添加环境变量示例及.gitignore更新

- 新增 .env.example,包含 Tushare API、钉钉机器人和PostgreSQL数据库配置模板
- 更新.gitignore,忽略本地配置文件如 .env.local 和 config_local.py
- 添加对报表文件命名规则的支持,保留示例文件不忽略
- 删除废弃的 chart.py 及相关图表模块代码
- 新增 config/settings.py,实现从环境变量读取配置的统一接口
- 设置数据目录及缓存目录,确保目录存在,提高配置管理规范性
This commit is contained in:
2026-03-18 23:33:40 +08:00
parent 7c93be4b41
commit 988c2335fb
39 changed files with 2983 additions and 1011 deletions

85
scripts/run_cci_screener.py Executable file
View File

@@ -0,0 +1,85 @@
#!/usr/bin/env python3
"""
CCI技术指标筛选器入口
用法:
python scripts/run_cci_screener.py
python scripts/run_cci_screener.py --config config/strategies/cci.yaml
python scripts/run_cci_screener.py --schedule # 定时模式
"""
import sys
import time
import yaml
import argparse
import schedule
from pathlib import Path
from datetime import datetime
# 添加项目根目录到路径
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
from strategies.screener.cci import CCIScreener
def load_config(config_path: str) -> dict:
"""加载配置文件"""
with open(config_path, "r", encoding="utf-8") as f:
return yaml.safe_load(f)
def run_screening(config: dict):
"""执行一次筛选"""
print(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 开始CCI筛选...")
screener = CCIScreener(config)
signals = screener.run_screening()
return signals
def main():
parser = argparse.ArgumentParser(description="CCI技术指标筛选")
parser.add_argument(
"--config",
type=str,
default="config/strategies/cci.yaml",
help="配置文件路径",
)
parser.add_argument(
"--schedule",
action="store_true",
help="启用定时模式",
)
args = parser.parse_args()
# 加载配置
config = load_config(args.config)
print("=" * 60)
print(" CCI技术指标筛选器")
print("=" * 60)
print(f"\n配置文件: {args.config}")
print(f"日线周期: {config.get('day_period', 14)}")
print(f"周线周期: {config.get('week_period', 14)}")
print(f"筛选阈值: {config.get('threshold', -100)}")
print(f"数据源: {config.get('data_source', 'postgresql')}")
if args.schedule:
# 定时模式
schedule_time = config.get("schedule_time", "19:00")
print(f"\n定时模式已启用,每天 {schedule_time} 执行")
print("按 Ctrl+C 停止\n")
schedule.every().day.at(schedule_time).do(run_screening, config)
while True:
schedule.run_pending()
time.sleep(1)
else:
# 单次执行
run_screening(config)
if __name__ == "__main__":
main()

108
scripts/run_rotation.py Executable file
View File

@@ -0,0 +1,108 @@
#!/usr/bin/env python3
"""
ETF轮动策略回测入口
用法:
python scripts/run_rotation.py
python scripts/run_rotation.py --config config/strategies/rotation.yaml
"""
import sys
import time
import yaml
import argparse
from pathlib import Path
# 添加项目根目录到路径
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
from strategies.rotation.engine import RotationStrategy
from strategies.rotation.portfolio import track_positions, save_trades
from strategies.rotation.report import generate_performance_report
from config.settings import CODE_NAME_MAP, BENCHMARK_NAME
def load_config(config_path: str) -> dict:
"""加载配置文件"""
with open(config_path, "r", encoding="utf-8") as f:
return yaml.safe_load(f)
def main():
parser = argparse.ArgumentParser(description="ETF轮动策略回测")
parser.add_argument(
"--config",
type=str,
default="config/strategies/rotation.yaml",
help="配置文件路径",
)
parser.add_argument(
"--save-path",
type=str,
default="report",
help="报告保存路径前缀",
)
args = parser.parse_args()
start_time = time.time()
print("=" * 60)
print(" ETF轮动策略 回测系统")
print("=" * 60)
# 加载配置
config = load_config(args.config)
print(f"\n配置文件: {args.config}")
print(f"候选标的: {len(config['code_list'])}")
print(f"回测区间: {config['start_date']} ~ {config['end_date']}")
print(f"因子类型: {config['factor_type']}")
print(f"窗口天数: {config['n_days']}")
print(f"选中数量: {config['select_num']}")
print(f"调仓周期: {config['rebalance_days']}")
print(f"交易成本: {config['trade_cost']:.2%}")
# 创建策略实例
strategy = RotationStrategy(config)
# 运行回测
print("\n" + "=" * 60)
print("开始回测...")
print("=" * 60)
backtest_result = strategy.run()
# 持仓跟踪
print("\n" + "=" * 60)
print("持仓跟踪...")
print("=" * 60)
trades_df, summary_df = track_positions(
backtest_result,
code_name_map=CODE_NAME_MAP,
select_num=config["select_num"],
)
save_trades(trades_df, summary_df, save_path=args.save_path)
# 生成绩效报告
print("\n" + "=" * 60)
print("生成绩效报告...")
print("=" * 60)
metrics = generate_performance_report(
backtest_result,
strategy.valid_codes,
code_name_map=CODE_NAME_MAP,
benchmark_name=BENCHMARK_NAME,
save_path=args.save_path,
select_num=config["select_num"],
)
elapsed = time.time() - start_time
print(f"\n总耗时: {elapsed:.1f}")
return metrics
if __name__ == "__main__":
main()