215 lines
5.1 KiB
Markdown
215 lines
5.1 KiB
Markdown
# 新架构说明文档
|
|
|
|
## 概述
|
|
|
|
本项目已重构为模块化的策略执行系统，支持定时执行多个策略任务。新架构分为四个层次：
|
|
|
|
1. **数据层**: TushareFetcher + StockPoolManager
|
|
2. **策略层**: BaseStrategy + 具体策略实现
|
|
3. **执行层**: TaskScheduler + StrategyExecutor
|
|
4. **通知层**: NotificationManager
|
|
|
|
## 核心工作流程
|
|
|
|
每个策略任务都遵循以下标准流程：
|
|
|
|
```
|
|
1. 根据规则获取股票池 → 2. 传递给策略进行分析 → 3. 以多种方式呈现或通知结果
|
|
```
|
|
|
|
## 模块详解
|
|
|
|
### 1. 数据层
|
|
|
|
#### TushareFetcher
|
|
- 负责从TuShare API获取股票数据
|
|
- 支持历史K线、热榜数据、基本信息等
|
|
- 已集成token管理和缓存机制
|
|
|
|
#### StockPoolManager
|
|
- 管理不同的股票池获取规则
|
|
- 内置规则：
|
|
- `tushare_hot`: 同花顺热榜
|
|
- `combined_hot`: 合并热门(同花顺+东财)
|
|
- `leading_stocks`: 龙头牛股
|
|
- 支持自定义股票列表
|
|
|
|
### 2. 策略层
|
|
|
|
#### BaseStrategy (抽象基类)
|
|
- 定义统一的策略接口
|
|
- 标准化输入输出格式
|
|
- 提供股票池批量分析能力
|
|
|
|
#### KLinePatternStrategy
|
|
- 实现K线形态识别策略
|
|
- 继承自BaseStrategy
|
|
- 支持多时间周期分析
|
|
- 集成回踩监控功能
|
|
|
|
### 3. 执行层
|
|
|
|
#### StrategyExecutor
|
|
- 协调股票池获取和策略分析
|
|
- 管理策略注册和执行
|
|
- 统一的结果处理和通知
|
|
|
|
#### TaskScheduler
|
|
- 支持多种调度规则
|
|
- 任务执行历史和统计
|
|
- 灵活的任务管理
|
|
|
|
### 4. 通知层
|
|
|
|
#### NotificationManager
|
|
- 支持多种通知方式
|
|
- 策略结果汇总推送
|
|
- 特殊事件提醒
|
|
|
|
## 快速开始
|
|
|
|
### 基础使用
|
|
|
|
```python
|
|
# 1. 初始化组件
|
|
from src.data.tushare_fetcher import TushareFetcher
|
|
from src.data.stock_pool_manager import StockPoolManager
|
|
from src.strategy.kline_pattern_strategy import KLinePatternStrategy
|
|
from src.execution.strategy_executor import StrategyExecutor
|
|
from src.utils.notification import NotificationManager
|
|
|
|
# 创建实例
|
|
fetcher = TushareFetcher()
|
|
pool_manager = StockPoolManager(fetcher)
|
|
notification_manager = NotificationManager(config)
|
|
executor = StrategyExecutor(pool_manager, notification_manager)
|
|
|
|
# 2. 注册策略
|
|
strategy = KLinePatternStrategy(fetcher, notification_manager, config)
|
|
executor.register_strategy("kline_pattern", strategy)
|
|
|
|
# 3. 执行任务
|
|
result = executor.execute_task(
|
|
task_id="test_task",
|
|
strategy_id="kline_pattern",
|
|
stock_pool_rule="tushare_hot",
|
|
max_stocks=20
|
|
)
|
|
```
|
|
|
|
### 定时任务
|
|
|
|
```python
|
|
from src.execution.task_scheduler import TaskScheduler
|
|
|
|
# 创建调度器
|
|
scheduler = TaskScheduler()
|
|
|
|
# 添加定时任务
|
|
task_func = executor.create_task_function(
|
|
strategy_id="kline_pattern",
|
|
stock_pool_rule="tushare_hot",
|
|
max_stocks=30
|
|
)
|
|
|
|
scheduler.add_task(
|
|
task_id="morning_scan",
|
|
name="晨间扫描",
|
|
func=task_func,
|
|
schedule_rule="weekdays at 09:00"
|
|
)
|
|
|
|
# 启动调度器
|
|
scheduler.start()
|
|
```
|
|
|
|
## 支持的调度规则
|
|
|
|
- `"every 30 minutes"` - 每30分钟
|
|
- `"every 1 hour"` - 每1小时
|
|
- `"daily at 09:30"` - 每日09:30
|
|
- `"weekdays at 14:00"` - 工作日14:00
|
|
- `"monday at 10:00"` - 每周一10:00
|
|
|
|
## 配置示例
|
|
|
|
### 策略配置
|
|
```python
|
|
strategy_config = {
|
|
'min_entity_ratio': 0.55,
|
|
'final_yang_min_ratio': 0.40,
|
|
'max_turnover_ratio': 40.0,
|
|
'timeframes': ['daily'],
|
|
'pullback_tolerance': 0.02,
|
|
'monitor_days': 30
|
|
}
|
|
```
|
|
|
|
### 股票池参数
|
|
```python
|
|
# 同花顺热榜
|
|
{"limit": 50}
|
|
|
|
# 合并热门
|
|
{"limit_per_source": 30, "final_limit": 50}
|
|
|
|
# 龙头股
|
|
{"top_boards": 8, "stocks_per_board": 3, "min_score": 70.0}
|
|
```
|
|
|
|
## 扩展指南
|
|
|
|
### 添加新的股票池规则
|
|
|
|
1. 继承 `StockPoolRule` 基类
|
|
2. 实现 `get_stocks()` 和 `get_rule_name()` 方法
|
|
3. 注册到 `StockPoolManager`
|
|
|
|
```python
|
|
class MyCustomRule(StockPoolRule):
|
|
def get_stocks(self, fetcher, **kwargs):
|
|
# 自定义获取逻辑
|
|
return stock_list
|
|
|
|
def get_rule_name(self):
|
|
return "我的自定义规则"
|
|
|
|
# 注册
|
|
pool_manager.register_rule("my_rule", MyCustomRule())
|
|
```
|
|
|
|
### 添加新策略
|
|
|
|
1. 继承 `BaseStrategy` 基类
|
|
2. 实现 `analyze_stock()` 和 `get_strategy_description()` 方法
|
|
3. 注册到 `StrategyExecutor`
|
|
|
|
```python
|
|
class MyStrategy(BaseStrategy):
|
|
def analyze_stock(self, stock_code, timeframes=None):
|
|
# 实现分析逻辑
|
|
return {timeframe: StrategyResult(...)}
|
|
|
|
def get_strategy_description(self):
|
|
return "我的策略描述"
|
|
|
|
# 注册
|
|
executor.register_strategy("my_strategy", MyStrategy(...))
|
|
```
|
|
|
|
## 示例文件
|
|
|
|
- `examples/new_architecture_example.py` - 完整架构演示
|
|
- `examples/task_config_examples.py` - 常见任务配置示例
|
|
|
|
## 优势总结
|
|
|
|
✅ **模块化设计**: 职责清晰，易于维护扩展
|
|
✅ **灵活配置**: 支持多种股票池和调度规则
|
|
✅ **标准化接口**: 统一的策略和结果格式
|
|
✅ **任务调度**: 强大的定时执行能力
|
|
✅ **统一通知**: 完善的结果推送机制
|
|
✅ **缓存优化**: 避免重复API调用
|
|
✅ **错误处理**: 完整的异常处理和日志记录
|
|
|
|
新架构完全满足了原始需求：通过简单配置实现复杂的多策略定时执行和结果通知系统。 |