# NSPS 周K线量化交易系统 — Function Specification
> Version: 1.0
> Created: 2026-07-30
> Status: Draft
---
## 模块总览
```
weekly-trading-system/
├── weekly_kline_layer.py # F1: 周K线量化分析
├── probability_engine.py # F2: 概率决策引擎
├── patient_trading_system.py # F3: 耐心交易系统
├── clfa_strategy_loop.py # F4: CLFA策略闭环
├── data/
│ ├── watchlist.json # 观察名单
│ ├── trades.json # 交易记录
│ └── weekly_stats.json # 周度统计
└── tests/
├── test_weekly_kline.py
├── test_probability.py
└── test_trading.py
```
---
## F1: weekly_kline_layer.py — 周K线量化分析
### 1.1 函数清单
```python
def get_weekly_ohlcv(symbol: str, market: str = 'US', weeks: int = 104) -> List[Dict]
"""
从TimescaleDB获取周K线数据
输入: symbol, market(US/CN), weeks(默认104=2年)
输出: [{date, open, high, low, close, volume}, ...]
逻辑: 将日线数据resample为周线（周五收盘）
"""
def detect_weekly_patterns(weekly_data: List[Dict]) -> Dict
"""
识别周K线形态
输入: 周K线数据
输出: {
"hammer": bool, # 锤子线
"engulfing": bool, # 吞没形态
"morning_star": bool, # 早晨之星
"three_white": bool, # 三连阳
"long_upper_shadow": bool, # 长上影线
"pattern_name": str, # 当前形态名称
"signal": str # BULLISH/BEARISH/NEUTRAL
}
"""
def calculate_weekly_indicators(weekly_data: List[Dict]) -> Dict
"""
计算周线技术指标
输出: {
"macd": {"value": float, "signal": float, "histogram": float, "cross": "golden"/"death"/None},
"rsi": float, # 14周RSI
"bollinger": {"upper": float, "middle": float, "lower": float, "position": "above_upper"/"below_lower"/"within"},
"ma_alignment": bool, # 5>10>20>50周MA多头排列
"ma50_slope": float, # 50周MA斜率（正=上升）
"above_ma50": bool # 当前价格是否在50周MA上方
}
"""
def detect_weekly_wave_phase(weekly_data: List[Dict]) -> Dict
"""
用周线判断Elliott波浪阶段
输出: {
"phase": "wave_1"/"wave_2"/"wave_3"/"wave_4"/"wave_5"/"unknown",
"confidence": float, # 0-1
"details": str # 描述
}
"""
def calculate_support_resistance(weekly_data: List[Dict]) -> Dict
"""
计算周线支撑/阻力位
输出: {
"resistance_levels": [float, ...], # 阻力位列表
"support_levels": [float, ...], # 支撑位列表
"fibonacci": {"382": float, "500": float, "618": float},
"52w_high": float,
"52w_low": float,
"distance_to_resistance": float, # 距最近阻力位%
"distance_to_support": float # 距最近支撑位%
}
"""
def calculate_relative_strength(symbol: str, market: str = 'US', weeks: int = 26) -> float
"""
计算相对强度排名（RS）
逻辑: 过去26周涨幅在全市场排名
输出: 0-100的百分位数（90=前10%）
"""
def weekly_kline_analysis(symbol: str, market: str = 'US') -> Dict
"""
主入口：完整周K线分析
输出: {
"symbol": str,
"market": str,
"timestamp": str,
"patterns": Dict, # from detect_weekly_patterns
"indicators": Dict, # from calculate_weekly_indicators
"wave_phase": Dict, # from detect_weekly_wave_phase
"support_resistance": Dict, # from calculate_support_resistance
"relative_strength": float, # from calculate_relative_strength
"composite_score": float, # 0-100综合评分
"signal": str # STRONG_BUY/BUY/HOLD/SELL/STRONG_SELL
}
"""
```
### 1.2 数据流
```
TimescaleDB (日线)
→ resample为周线
→ 形态识别 + 技术指标 + 波浪位置 + 支撑阻力 + RS
→ 综合评分
→ 输出信号
```
### 1.3 依赖
- `db_utils.py` — 数据库连接
- `numpy` — 数值计算
- 无外部API依赖（纯本地计算）
---
## F2: probability_engine.py — 概率决策引擎
### 2.1 函数清单
```python
def calculate_expected_value(win_rate: float, avg_win: float, avg_loss: float) -> float
"""
计算期望值
E = win_rate × avg_win - (1-win_rate) × avg_loss
"""
def kelly_criterion(win_rate: float, payoff_ratio: float, fraction: float = 0.5) -> float
"""
凯利公式（默认半凯利）
f = (win_rate × payoff_ratio - 1) / (payoff_ratio - 1) × fraction
"""
def calculate_position_size(capital: float, entry: float, stop_loss: float,
max_risk_pct: float = 0.01) -> float
"""
1%风险原则计算仓位
position = (capital × max_risk_pct) / (entry - stop_loss)
"""
def backtest_pattern(pattern_name: str, market: str = 'US',
lookback_weeks: int = 520) -> Dict
"""
回测特定形态的历史胜率
输入: 形态名称，市场，回看周数
输出: {
"total_occurrences": int,
"win_rate": float,
"avg_win": float,
"avg_loss": float,
"expected_value": float,
"max_consecutive_losses": int,
"payoff_ratio": float
}
"""
def update_strategy_parameters(trades: List[Dict]) -> Dict
"""
根据最近50笔交易自动调整策略参数
逻辑:
- 胜率<40% → 收紧入场条件
- 盈亏比<1.5 → 调整止损/止盈
- 某形态连续失败3次 → 降权
"""
def probability_assessment(symbol: str, pattern: Dict, indicators: Dict) -> Dict
"""
概率评估（主入口）
输出: {
"pattern_name": str,
"historical_win_rate": float,
"expected_value": float,
"kelly_position": float, # 凯利建议仓位
"risk_adjusted_position": float, # 1%原则仓位
"recommended_position": float, # 最终建议仓位
"signal_strength": str # STRONG/MODERATE/WEAK
}
"""
```
### 2.2 数据流
```
历史形态数据 (TimescaleDB)
→ 回测统计胜率/盈亏比
→ 期望值计算
→ 凯利公式 → 仓位建议
→ 1%风险原则 → 最终仓位
```
---
## F3: patient_trading_system.py — 耐心交易系统
### 3.1 函数清单
```python
def screen_leaders(market: str = 'US') -> List[Dict]
"""
量化筛选清单（A+B+C条件）
A: 股价>50周MA 且 MA斜率>0
B: RS前10%
C: 营收增长>20%（如有数据）
输出: 符合条件的候选股列表
"""
def update_watchlist(candidates: List[Dict]) -> List[Dict]
"""
更新观察名单
逻辑: 新候选加入，已触发/过期的移除
输出: 更新后的观察名单
"""
def check_trigger_conditions(symbol: str) -> Dict
"""
检查触发条件（至少3/5满足）
1. 周线形态：锤子线/吞没/早晨之星
2. 周线指标：MACD金叉 + RSI<50
3. 周线波浪：2浪末期或3浪初期
4. 周线成交量：放量突破
5. 周线支撑/阻力：突破关键阻力
输出: {
"conditions_met": int,
"conditions_detail": [...],
"triggered": bool # conditions_met >= 3
}
"""
def calculate_entry_plan(symbol: str, capital: float, trigger: Dict) -> Dict
"""
计算入场计划
输出: {
"entry_price": float,
"stop_loss": float,
"target_price": float,
"position_size": float, # 1%风险原则
"tranche_1": float, # 试探仓25%
"tranche_2": float, # 加仓仓25%
"tranche_3": float, # 锁定仓25%
"max_loss": float, # 最大亏损金额
"risk_reward_ratio": float # 盈亏比
}
"""
def manage_position(position: Dict, current_price: float) -> Dict
"""
持仓管理
逻辑:
- 周线趋势未坏 + 未触及止损 → 持有
- 盈利达到目标 → 考虑加仓
- 触及止损 → 无条件退出
- 亏损头寸 → 绝不加仓
输出: {
"action": "HOLD"/"ADD"/"EXIT",
"reason": str,
"updated_stop_loss": float
}
"""
def weekly_review() -> Dict
"""
每周复盘（每周日运行）
输出: {
"trades_this_week": int,
"pnl_this_week": float,
"current_positions": [...],
"watchlist_changes": [...],
"strategy_adjustments": [...]
}
"""
```
### 3.2 状态机
```
SCREENING → WATCHLIST → TRIGGERED → ENTRY → MANAGING → EXIT
↓ ↑
└── NOT_TRIGGERED ──────────┘
(继续等待)
```
---
## F4: clfa_strategy_loop.py — CLFA策略闭环
### 4.1 函数清单
```python
def record_trade(trade: Dict) -> None
"""
记录交易（6个数据点）
trade: {symbol, entry_price, stop_loss, target_price,
exit_price, weeks_held, result}
"""
def weekly_backtest(lookback: int = 50) -> Dict
"""
每周自动回测最近N笔
输出: {
"win_rate": float,
"avg_win_loss_ratio": float,
"expected_value": float,
"failing_patterns": [...], # 连续失败3次的形态
"parameter_adjustments": [...]
}
"""
def auto_adjust_parameters(backtest_result: Dict) -> Dict
"""
自动调整策略参数
逻辑:
- 胜率<40% → 收紧入场条件
- 盈亏比<1.5 → 调整止损/止盈
- 某形态连续失败3次 → 降权/禁用
"""
```
---
## 独立性保证
| 原则 | 实现方式 |
|------|----------|
| 不影响现有系统 | 独立目录 `weekly-trading-system/` |
| 数据共享 | 共用TimescaleDB，只读不写 |
| 渐进集成 | 每个模块独立测试，通过后才集成 |
| 向后兼容 | 现有日K线Pipeline继续运行 |
| 配置独立 | 独立配置文件 `weekly_config.json` |
---
*Last updated: 2026-07-30 14:50 PT*