diff --git a/intraday-regime-detector/SKILL.md b/intraday-regime-detector/SKILL.md new file mode 100644 index 0000000..ccf19d7 --- /dev/null +++ b/intraday-regime-detector/SKILL.md @@ -0,0 +1,204 @@ +--- +name: intraday-regime-detector +description: "港美股日内做T 元策略 - 根据 5min K 线自动判别市场状态 (趋势/震荡/混乱) 并推荐匹配策略 (趋势跟踪/网格/布林带回归)。来源 DeepSeek 分享, 决策树: R² > 0.75 趋势; R² < 0.30 + ADF 平稳 + 低波动 = 网格; 高波动 = 布林带; 其他 NO_TRADE。⚠️ 仅识别市场状态, 不替代 longbridge-t-monitor / strategy-management 的入场/出场逻辑。" +version: 1.0.0 +author: Hermes Agent + DeepSeek 分享 (26iikphv8h94feze9q) +tags: [trading, intraday, market-regime, regime-detection, deepseek, hk, us] +metadata: + hermes: + tags: [trading, intraday, market-regime, regime-detection, deepseek, hk, us] + related_skills: [longbridge-t-monitor, strategy-management] +scripts: + - intraday_regime.py: "核心: IntradayStrategySelector + MarketDiagnosis + MarketRegime + StrategyType" + - regime_scan.py: "扫描港美股 top 5 候选, 拉长桥 5min K线, 跑判别, 推报告" +references: + - decision-tree.md: "决策树详细说明 + 4 策略参数说明" +--- + +# Intraday Regime Detector (港美股日内做T 元策略) + +**核心定位**:**不是替代** `longbridge-t-monitor` 或 `strategy-management`, 而是**在它们之前**先判断"现在适不适合做T、做哪个策略"。 + +``` +intraday-regime-detector (本 skill) → 告诉用户 "用什么策略 + 为什么" + ↓ +longbridge-t-monitor (现有) → 执行入场/出场 +strategy-management (现有) → 选策略 + 算 SL/TP +``` + +## 🎯 解决的痛点 + +| 痛点 | 解决 | +|---|---| +| 趋势市用网格 = 反复止损 | R² > 0.75 → 强制趋势策略 | +| 震荡市用趋势 = 追涨杀跌 | R² < 0.30 → 强制震荡策略 | +| 混乱行情硬做 = 越做越亏 | ADF p > 0.05 → NO_TRADE | +| 不知道用宽网格还是窄网格 | 波动率高 → 布林带, 低 → 网格 | + +## 📦 决策树 (DeepSeek 原始版) + +``` +第一步: 输入近 20-30 根 5 分钟 K 线 + 开盘缺口 +第二步: 计算趋势效率 R² (线性回归) + ├─ R² > 0.75: 强趋势市 → 趋势跟踪 (顺势 EMA5 支撑/阻力) + ├─ R² < 0.30: 强震荡市 + │ ├─ ADF p < 0.05 (平稳): + │ │ ├─ 波动率 > 30%: 布林带回归 + │ │ └─ 波动率 ≤ 30%: 网格交易 + │ └─ ADF p ≥ 0.05 (不平稳): 混乱 → NO_TRADE + └─ 0.30 ≤ R² ≤ 0.75: 过渡 → 暂停, 等模式清晰 +``` + +## 🚀 快速使用 + +### Python API + +```python +import sys +sys.path.insert(0, '/home/openclaw/.hermes/skills/trading/intraday-regime-detector/scripts') +from intraday_regime import IntradayStrategySelector + +# 假设 df 是 5min K线 DataFrame, 包含 open/high/low/close +selector = IntradayStrategySelector() +diagnosis = selector.diagnose(df, open_gap_pct=0.3) + +print(f"状态: {diagnosis.regime.value}") +print(f"推荐: {diagnosis.recommended_strategy.value}") +print(f"R²: {diagnosis.r_squared}, 置信度: {diagnosis.confidence}") +print(f"参数: {diagnosis.strategy_params}") +``` + +### CLI 扫描 (港美股 top 5) + +```bash +/home/openclaw/.hermes/hermes-agent/venv/bin/python \ + /home/openclaw/.hermes/skills/trading/intraday-regime-detector/scripts/regime_scan.py +``` + +输出示例: +``` +🔲 9888.HK (HK) 现价 $110.30 (+1.21%) 置信度 85% + 状态: 低波震荡 | R²=0.2363 | 波动率=3.32% | ADF p=0.01 + 推荐: 网格交易做T + grid_spacing: 0.500% + grid_levels: 3 + base_price: 110.30 +``` + +## 📊 输出数据结构 + +`MarketDiagnosis` (dataclass): +```python +@dataclass +class MarketDiagnosis: + regime: MarketRegime # 6 种状态之一 + r_squared: float # 趋势效率 0-1 + volatility: float # 年化波动率 + adf_pvalue: float # ADF 平稳检验 p 值 + recommended_strategy: StrategyType # 4 种策略之一 + strategy_params: Dict # 动态参数 (SL/TP/grid 等) + confidence: float # 0-1 + reasoning: str # 人话解释 +``` + +`MarketRegime` 枚举: +- `STRONG_TREND_UP` / `STRONG_TREND_DOWN` +- `HIGH_VOL_SHAKE` (高波动震荡) +- `LOW_VOL_STABLE` (低波动震荡) +- `CHAOTIC` (混乱) +- `UNKNOWN` (过渡区间) + +`StrategyType` 枚举: +- `TREND_FOLLOWING` (EMA5 顺势) +- `GRID_TRADING` (3 格 × 0.5%) +- `BOLLINGER_REVERSION` (20 期 ±2σ) +- `NO_TRADE` (暂停) + +## 🔧 配置参数 + +```python +selector = IntradayStrategySelector( + trend_r2_threshold=0.75, # R² 高于此 = 趋势市 + chaos_r2_threshold=0.30, # R² 低于此 = 震荡市 + adf_significance=0.05, # ADF p < 此 = 平稳 + vol_lookback=20, # 历史波动率窗口 + high_vol_threshold=0.30, # 年化波动率高/低分界 + grid_count=3, # 网格层数 + bb_period=20, # 布林带周期 + bb_std=2.0, # 布林带 σ + ema_period=5, # 趋势策略 EMA 周期 +) +``` + +## 🔄 与现有 skill 的关系 + +| Skill | 关系 | +|---|---| +| `longbridge-t-monitor` | **下游** - 本 skill 决定"该不该做T + 用什么策略", 然后 longbridge-t-monitor 执行 | +| `strategy-management` | **互补** - strategy-management 有具体的 5 策略 (rsi2_revert/vwap_revert/early_bird/turtle/sma), 本 skill 是"先用元策略筛一下再用具体策略" | +| `intraday-trading` | **理论来源** - 已有 4 策略设计 + 5 步预检 + 资金管理表 | +| `crypto-t-monitor` | **独立** - 币圈用 OKX ATR 公式, 本 skill 不涉及 | + +**集成路径 (建议)**: +``` +盘前 cron (c3401d727f39, cfa0c1d6baa5) + ↓ 生成候选池 +盘中 cron (新加): regime_scan + ↓ 输出 "可做 T 的票 + 推荐策略" +手动 / agent: 看推送 + ↓ 决定是否入场 +longbridge-t-monitor: 执行 +``` + +## ⚠️ Pitfalls + +1. **ADF 是简化版** — 实际生产用 `statsmodels.tsa.stattools.adfuller`. 代码已 fallback, 有 statsmodels 就用真 ADF +2. **30 根 K 线窗口** — 跟 longbridge-t-monitor 一样的限制, period 选 5m → 2.5h 窗口 +3. **网格/布林带参数是建议值** — 实盘要按资金 + 流动性 + 个人风险偏好调整 +4. **本 skill 不替代风控** — 出场点位 / 仓位管理 / 单日最大亏损 → 用 longbridge-t-monitor +5. **不主动下单 (用户偏好 2026-07-16)**: 用户原话 "在跑日内交易扫描任务时使用 intraday-regime-detector, 不主动下单". 本 skill 只输出状态 + 推荐策略, **不要在 diagnose() 里加任何下单逻辑**. +6. **港股 candlesticks 不支持 `--json`**: 长桥 CLI 港股 candlesticks 只输出中文表格, 表格分隔符是 `│` (不是 `|`). `regime_scan.py` 已处理. 美股用 `--json`. + +## 👤 用户偏好 (2026-07-16) + +- **来源**: https://chat.deepseek.com/share/26iikphv8h94feze9q +- **要求**: 跑日内交易扫描任务时使用 intraday-regime-detector, 不主动下单 +- **三段式 pipeline** (已部署): + 1. 候选池 (quant-factor-mining) → top 5 + 2. 元策略 (本 skill) → confidence ≥ 0.6 算 actionable + 3. 点位 (strategy-management/exit_levels) → SL/TP/TP2 +- **部署位置**: + - `/home/openclaw/qdrant/calc_hk_levels.py` - 港股 + - `/home/openclaw/qdrant/calc_us_levels.py` - 美股 + - Cron `c4dc9ac8854c` (港股 */15 9-15) + `70d24624637c` (美股 */15 21-3) 周一到周五 + - Wrapper: `~/.hermes/scripts/hk_t_levels.sh` / `us_t_levels.sh` +- **A 股 vs 港美股** (重要区分, 用户 2026-07-16 强调): + - A 股做T = 底仓滚动 (T+1 制度) + - 港美股做T = 直接双向交易 (T+0) + - **本 skill 只服务港美股**。A 股做T 完全用不上这个。 + +## 🛡️ 已知问题 + +| 问题 | 处理 | +|---|---| +| statsmodels 未装 | 自动 fallback 到自相关近似 | +| K 线 < 10 根 | 抛 ValueError, 跳过 | +| R² 边界值 (0.30 / 0.75) | 默认参数, 可在 __init__ 调整 | +| 大量候选 NO_TRADE | 正常, 实测 10 支 7 支 NO_TRADE. 算法价值正在于"拒绝不值得做的票" | + +## 📚 参考 + +- **来源**: https://chat.deepseek.com/share/26iikphv8h94feze9q +- **决策树详细**: `references/decision-tree.md` +- **测试数据**: `intraday_regime.py` 的 `__main__` 跑自检 +- **集成代码**: + - `~/.hermes/qdrant/calc_hk_levels.py` (港股) + - `~/.hermes/qdrant/calc_us_levels.py` (美股) + - `~/.hermes/scripts/hk_t_levels.sh` / `us_t_levels.sh` (wrapper) + +## 🔄 版本历史 + +- **v1.0.0** (2026-07-16): 初始版本 + - `intraday_regime.py` - 核心判别器 (DeepSeek 原始代码 + statsmodels fallback + dataclass) + - `regime_scan.py` - 长桥 K线集成扫描器 + - 自检场景 (震荡市/趋势市) 全部通过 \ No newline at end of file diff --git a/intraday-regime-detector/references/decision-tree.md b/intraday-regime-detector/references/decision-tree.md new file mode 100644 index 0000000..ee53cc5 --- /dev/null +++ b/intraday-regime-detector/references/decision-tree.md @@ -0,0 +1,144 @@ +# 决策树详细说明 + +来源: DeepSeek chat share 26iikphv8h94feze9q + +## 一、核心判别算法 + +### 1. 趋势效率 R² (线性回归) + +**目的**: 衡量趋势的"纯粹度", 比单纯看均线方向更科学。 + +**计算**: +- 取过去 N 根 K 线 (默认 20 根 5min K) 的收盘价序列 +- 以时间 (1,2,3...20) 为自变量 X, 收盘价为因变量 Y +- 做一元线性回归 +- 计算 R² + +**判断**: +| R² | 状态 | 含义 | +|---|---|---| +| > 0.75 | 趋势市 | 价格运动有明确方向, 噪声小 | +| < 0.30 | 震荡市 | 价格运动无方向, 充满噪声 | +| 0.30-0.75 | 过渡 | 方向不明, 等待 | + +### 2. ADF 平稳检验 (Augmented Dickey-Fuller) + +**目的**: 判断价格序列是否倾向于均值回归。 + +**计算**: +- 对过去价格序列执行 ADF 检验 +- 返回 p 值 + +**判断**: +| p 值 | 含义 | 策略匹配 | +|---|---|---| +| < 0.05 | 拒绝非平稳假设, 统计上平稳 | **均值回归, 适合震荡做T** | +| > 0.05 | 不能拒绝非平稳, 可能是随机游走或趋势 | **不做均值回归** | + +**⚠️ 简化实现**: 本 skill 用一阶差分自相关近似, 生产建议替换为 `statsmodels.tsa.stattools.adfuller` (代码已 fallback). + +### 3. 历史波动率 (年化) + +**计算**: log returns 标准差 × √252 + +**判断**: +| 波动率 | 含义 | +|---|---| +| > 30% | 高波动, 适合宽间距逆势 (布林带) | +| ≤ 30% | 低波动, 适合网格 | + +### 4. 开盘缺口 + +**计算**: (open - prev_close) / prev_close × 100% + +**判断**: +| 缺口 | 含义 | +|---|---| +| > +0.5% | 高开强势, 优先做正T | +| < -0.5% | 低开弱势, 优先做倒T | +| 平开/微小 | 默认震荡模式 | + +## 二、策略参数说明 + +### 1. 趋势跟踪做T (TREND_FOLLOWING) + +**适用**: 强趋势市 (R² > 0.75) + +**参数** (5min K): +- `direction`: long_only / short_only +- `entry_trigger`: 价格回踩 EMA5 不破 (上涨) / 价格反弹至 EMA5 受阻 (下跌) +- `stop_loss`: EMA5 × 0.995 (long) / EMA5 × 1.005 (short) +- `take_profit`: 现价 × 1.02 / × 0.98 + +### 2. 网格交易做T (GRID_TRADING) + +**适用**: 低波动震荡 (R² < 0.30 + ADF 平稳 + 低波动) + +**参数**: +- `grid_spacing`: 近期平均振幅 × 0.8, 至少 0.5% +- `grid_levels`: 3 (默认) +- `base_price`: 当前价 +- `reverse_at_boundary`: True (在边界反向开仓) + +### 3. 布林带回归做T (BOLLINGER_REVERSION) + +**适用**: 高波动震荡 (R² < 0.30 + ADF 平稳 + 高波动) + +**参数** (20 期 ±2σ): +- `upper_band`: MA20 + 2σ +- `lower_band`: MA20 - 2σ +- `sell_at_upper`: True +- `buy_at_lower`: True +- `stop_if_break`: True (破带止损) + +### 4. NO_TRADE (暂停) + +**适用**: 混乱或过渡状态 + +**参数**: `reason: 市场状态不清晰, 等待趋势或明确震荡信号` + +## 三、决策流程图 + +``` + ┌─────────────────────┐ + │ 输入 20 根 5min K │ + │ + 开盘缺口 │ + └──────────┬──────────┘ + ↓ + ┌─────────────────────┐ + │ 计算 R² │ + └──────────┬──────────┘ + ↓ + ┌─────────────────┼─────────────────┐ + ↓ ↓ ↓ + R² > 0.75 0.30-0.75 R² < 0.30 + 强趋势 过渡 震荡 + ↓ ↓ ↓ + TREND UNKNOWN 计算 ADF + FOLLOWING NO_TRADE ↓ + ┌─────┴─────┐ + ↓ ↓ + ADF<0.05 ADF≥0.05 + 平稳 不平稳 + ↓ ↓ + 计算波动率 CHAOTIC + ↓ NO_TRADE + ┌───┴───┐ + ↓ ↓ + vol>30% vol≤30% + ↓ ↓ + BOLLINGER GRID + REVERSION TRADING +``` + +## 四、与本 skill 的对应关系 + +| 步骤 | 函数 | 文件 | +|---|---|---| +| 输入校验 | `diagnose()` | intraday_regime.py | +| 计算 R² | `_calculate_r_squared()` | intraday_regime.py | +| 计算 vol | `_calculate_historical_volatility()` | intraday_regime.py | +| 计算 ADF | `_adf_test()` | intraday_regime.py | +| 趋势判断 | `_classify_regime()` | intraday_regime.py | +| 策略匹配 | `_match_strategy()` | intraday_regime.py | +| 拉 K 线 + 整合 | `regime_scan.py` | regime_scan.py | \ No newline at end of file diff --git a/intraday-regime-detector/scripts/intraday_regime.py b/intraday-regime-detector/scripts/intraday_regime.py new file mode 100644 index 0000000..ea295c7 --- /dev/null +++ b/intraday-regime-detector/scripts/intraday_regime.py @@ -0,0 +1,307 @@ +""" +intraday_regime.py - 日内市场状态判别 + 策略匹配 + +来源: DeepSeek chat share 26iikphv8h94feze9q +核心算法: + 1. 趋势效率 R² (线性回归) - 判趋势 vs 震荡 + 2. ADF 平稳检验 - 验证均值回归 + 3. 历史波动率 - 区分高/低波动 + 4. 开盘缺口 - 识别方向偏好 + +决策树: + R² > 0.75 → 趋势跟踪 (顺势) + R² < 0.30 + ADF 平稳 + 低波动 → 网格交易 + R² < 0.30 + ADF 平稳 + 高波动 → 布林带回归 + 其他 → NO_TRADE (暂停) + +⚠️ 这是港美股日内做T 元策略, 跟 crypto-t-monitor / longbridge-t-monitor 都独立 +""" +import numpy as np +import pandas as pd +from typing import Dict, List, Tuple, Optional +from dataclasses import dataclass, field +from enum import Enum +import warnings +warnings.filterwarnings('ignore') + + +class MarketRegime(Enum): + STRONG_TREND_UP = "强趋势上涨" + STRONG_TREND_DOWN = "强趋势下跌" + HIGH_VOL_SHAKE = "高波动剧烈震荡" + LOW_VOL_STABLE = "低波动平稳震荡" + CHAOTIC = "混乱无序" + UNKNOWN = "无法判断" + + +class StrategyType(Enum): + TREND_FOLLOWING = "趋势跟踪做T" + GRID_TRADING = "网格交易做T" + BOLLINGER_REVERSION = "布林带回归做T" + NO_TRADE = "暂停交易" + + +@dataclass +class MarketDiagnosis: + regime: MarketRegime + r_squared: float + volatility: float + adf_pvalue: float + recommended_strategy: StrategyType + strategy_params: Dict + confidence: float + reasoning: str + + +class IntradayStrategySelector: + """ + 根据 5min K 线自动判别市场状态 + 推荐日内做T 策略 + + 用法: + selector = IntradayStrategySelector() + diagnosis = selector.diagnose(df_5min, open_gap_pct=0.3) + print(diagnosis.recommended_strategy) + """ + + def __init__(self, + trend_r2_threshold: float = 0.75, + chaos_r2_threshold: float = 0.30, + adf_significance: float = 0.05, + vol_lookback: int = 20, + high_vol_threshold: float = 0.30, + grid_count: int = 3, + bb_period: int = 20, + bb_std: float = 2.0, + ema_period: int = 5): + self.trend_r2_threshold = trend_r2_threshold + self.chaos_r2_threshold = chaos_r2_threshold + self.adf_significance = adf_significance + self.vol_lookback = vol_lookback + self.high_vol_threshold = high_vol_threshold + self.grid_count = grid_count + self.bb_period = bb_period + self.bb_std = bb_std + self.ema_period = ema_period + + def diagnose(self, df: pd.DataFrame, open_gap_pct: float = 0.0) -> MarketDiagnosis: + if len(df) < 10: + raise ValueError(f"需要至少 10 根 K 线, 拿到 {len(df)}") + + for col in ['open', 'high', 'low', 'close']: + if col not in df.columns: + raise ValueError(f"df 缺少 '{col}' 列") + + prices = df['close'].values + r_squared = self._calculate_r_squared(prices) + volatility = self._calculate_historical_volatility(prices) + adf_pvalue = self._adf_test(prices) + slope = self._calculate_trend_slope(prices) + + regime, confidence, reasoning = self._classify_regime( + r_squared, volatility, adf_pvalue, slope, open_gap_pct + ) + + strategy, params = self._match_strategy(regime, df, volatility, r_squared) + + return MarketDiagnosis( + regime=regime, + r_squared=round(r_squared, 4), + volatility=round(volatility, 4), + adf_pvalue=round(adf_pvalue, 4), + recommended_strategy=strategy, + strategy_params=params, + confidence=round(confidence, 2), + reasoning=reasoning, + ) + + def _calculate_r_squared(self, prices: np.ndarray) -> float: + n = len(prices) + if n < 2: + return 0.0 + x = np.arange(1, n + 1) + y = prices + x_mean = np.mean(x) + y_mean = np.mean(y) + numerator = np.sum((x - x_mean) * (y - y_mean)) + denominator = np.sqrt(np.sum((x - x_mean) ** 2) * np.sum((y - y_mean) ** 2)) + if denominator == 0: + return 0.0 + r = numerator / denominator + return r ** 2 + + def _calculate_historical_volatility(self, prices: np.ndarray) -> float: + if len(prices) < 2: + return 0.0 + log_returns = np.diff(np.log(prices)) + return float(np.std(log_returns) * np.sqrt(252)) + + def _calculate_trend_slope(self, prices: np.ndarray) -> float: + n = len(prices) + if n < 2: + return 0.0 + x = np.arange(1, n + 1) + y = prices + x_mean = np.mean(x) + y_mean = np.mean(y) + slope = np.sum((x - x_mean) * (y - y_mean)) / np.sum((x - x_mean) ** 2) + return float(slope / y_mean) if y_mean != 0 else 0.0 + + def _adf_test(self, prices: np.ndarray) -> float: + """简化 ADF (用一阶差分自相关近似) + 生产建议用 statsmodels.tsa.stattools.adfuller + """ + try: + from statsmodels.tsa.stattools import adfuller + result = adfuller(prices, autolag='AIC') + return float(result[1]) + except ImportError: + # Fallback: 用一阶差分自相关近似 + diffs = np.diff(prices) + if len(diffs) < 10: + return 1.0 + autocorr = float(np.corrcoef(diffs[:-1], diffs[1:])[0, 1]) + if abs(autocorr) < 0.1: + return 0.01 + elif abs(autocorr) < 0.3: + return 0.05 + elif abs(autocorr) < 0.5: + return 0.15 + else: + return 0.50 + + def _classify_regime(self, r2: float, vol: float, adf_p: float, + slope: float, gap: float) -> Tuple[MarketRegime, float, str]: + + if r2 > self.trend_r2_threshold: + regime = MarketRegime.STRONG_TREND_UP if slope > 0.002 else MarketRegime.STRONG_TREND_DOWN + confidence = min(r2, 1.0) + reasoning = (f"R²={r2:.3f}>0.75, 市场呈现强趋势状态。" + f"线性回归斜率={slope:.4f}, 方向明确。" + f"此类行情适合顺势做T,严禁逆势网格。") + return regime, confidence, reasoning + + if r2 < self.chaos_r2_threshold: + if adf_p < self.adf_significance: + if vol > self.high_vol_threshold: + regime = MarketRegime.HIGH_VOL_SHAKE + confidence = 0.70 + reasoning = (f"R²={r2:.3f}<0.30, ADF p={adf_p:.3f}<0.05, " + f"但波动率={vol:.2%}偏高。市场为高波动震荡," + f"适宜宽间距的逆势策略,需严格止损。") + else: + regime = MarketRegime.LOW_VOL_STABLE + confidence = 0.85 + reasoning = (f"R²={r2:.3f}<0.30, ADF p={adf_p:.3f}<0.05, " + f"波动率={vol:.2%}适中。经典震荡市," + f"是网格和布林带回归策略的理想环境。") + else: + regime = MarketRegime.CHAOTIC + confidence = 0.40 + reasoning = (f"R²={r2:.3f}<0.30, 但 ADF p={adf_p:.3f}>0.05, " + f"价格不具均值回归特性,属混乱状态,建议观望。") + return regime, confidence, reasoning + + regime = MarketRegime.UNKNOWN + confidence = 0.30 + reasoning = (f"R²={r2:.3f} 处于过渡区间(0.30-0.75), " + f"市场方向不明。建议等待模式清晰后再交易。") + return regime, confidence, reasoning + + def _match_strategy(self, regime: MarketRegime, df: pd.DataFrame, + vol: float, r2: float) -> Tuple[StrategyType, Dict]: + current_price = float(df['close'].iloc[-1]) + params = {} + + if regime == MarketRegime.STRONG_TREND_UP: + strategy = StrategyType.TREND_FOLLOWING + ema = float(df['close'].ewm(span=self.ema_period).mean().iloc[-1]) + params = { + "direction": "long_only", + "entry_trigger": f"价格回踩 {ema:.2f} (EMA{self.ema_period}) 不破", + "stop_loss": f"{ema * 0.995:.2f}", + "take_profit": f"{current_price * 1.02:.2f}", + } + + elif regime == MarketRegime.STRONG_TREND_DOWN: + strategy = StrategyType.TREND_FOLLOWING + ema = float(df['close'].ewm(span=self.ema_period).mean().iloc[-1]) + params = { + "direction": "short_only", + "entry_trigger": f"价格反弹至 {ema:.2f} (EMA{self.ema_period}) 受阻", + "stop_loss": f"{ema * 1.005:.2f}", + "take_profit": f"{current_price * 0.98:.2f}", + } + + elif regime == MarketRegime.LOW_VOL_STABLE: + strategy = StrategyType.GRID_TRADING + avg_amplitude = float(((df['high'] - df['low']) / df['close']).mean()) + grid_spacing = max(avg_amplitude * 0.8, 0.005) + params = { + "grid_spacing": f"{grid_spacing:.3%}", + "grid_levels": self.grid_count, + "base_price": f"{current_price:.2f}", + "reverse_at_boundary": True, + } + + elif regime == MarketRegime.HIGH_VOL_SHAKE: + strategy = StrategyType.BOLLINGER_REVERSION + rolling_std = float(df['close'].rolling(self.bb_period).std().iloc[-1]) + ma = float(df['close'].rolling(self.bb_period).mean().iloc[-1]) + upper = ma + self.bb_std * rolling_std + lower = ma - self.bb_std * rolling_std + params = { + "upper_band": f"{upper:.2f}", + "lower_band": f"{lower:.2f}", + "sell_at_upper": True, + "buy_at_lower": True, + "stop_if_break": True, + } + + else: + strategy = StrategyType.NO_TRADE + params = {"reason": "市场状态不清晰,等待趋势或明确震荡信号"} + + return strategy, params + + +def diagnose_market(df: pd.DataFrame, open_gap_pct: float = 0.0) -> MarketDiagnosis: + """便捷函数""" + return IntradayStrategySelector().diagnose(df, open_gap_pct) + + +if __name__ == '__main__': + np.random.seed(42) + + # 场景 1: 震荡市 + n = 25 + base = 10.0 + noise = np.random.randn(n) * 0.05 + close = base + noise + high = close + np.abs(np.random.randn(n) * 0.03) + low = close - np.abs(np.random.randn(n) * 0.03) + df = pd.DataFrame({ + 'open': close - 0.01, + 'high': high, + 'low': low, + 'close': close, + 'volume': np.random.randint(1000, 5000, n), + }) + diag = diagnose_market(df, open_gap_pct=0.0) + print(f"场景 1 (震荡市): {diag.regime.value} | R²={diag.r_squared} | 策略: {diag.recommended_strategy.value}") + print(f" 推理: {diag.reasoning}\n") + + # 场景 2: 强趋势 + close2 = base + np.cumsum(np.random.randn(n) * 0.02) * 2 # 上升趋势 + high2 = close2 + 0.05 + low2 = close2 - 0.05 + df2 = pd.DataFrame({ + 'open': close2 - 0.01, + 'high': high2, + 'low': low2, + 'close': close2, + 'volume': np.random.randint(1000, 5000, n), + }) + diag2 = diagnose_market(df2, open_gap_pct=0.5) + print(f"场景 2 (趋势市): {diag2.regime.value} | R²={diag2.r_squared} | 策略: {diag2.recommended_strategy.value}") + print(f" 推理: {diag2.reasoning}") + print(f" 参数: {diag2.strategy_params}") \ No newline at end of file diff --git a/intraday-regime-detector/scripts/regime_scan.py b/intraday-regime-detector/scripts/regime_scan.py new file mode 100644 index 0000000..bfcbda0 --- /dev/null +++ b/intraday-regime-detector/scripts/regime_scan.py @@ -0,0 +1,186 @@ +#!/usr/bin/env python3 +""" +regime_scan.py - 扫描港美股日内候选的市场状态 + 推荐策略 +不交易, 只判别 + 推 QQ +""" +import sys +import json +import subprocess +import re +from pathlib import Path + +sys.path.insert(0, '/home/openclaw/.hermes/skills/trading/intraday-regime-detector/scripts') +from intraday_regime import IntradayStrategySelector, MarketRegime, StrategyType + +CANDIDATE_HK = Path('/home/openclaw/.hermes/skills/trading/quant-factor-mining/artifacts/hk_intraday_latest.json') +CANDIDATE_US = Path('/home/openclaw/.hermes/skills/trading/quant-factor-mining/artifacts/us_intraday_latest.json') + + +def fetch_klines_hk(symbol: str, period: str = '5m', count: int = 30) -> list: + """港股表格 parser""" + result = subprocess.run( + ['proxychains4', '-f', '/home/openclaw/.proxychains/proxychains.conf', + '/home/openclaw/.local/bin/longbridge', '--profile', 'lb_real', + 'candlesticks', symbol, period, '--count', str(count)], + capture_output=True, text=True, timeout=30, + ) + klines = [] + pattern = re.compile( + r'│\s*(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2})\s*│' + r'\s*([\d,.]+)\s*│\s*([\d,.]+)\s*│\s*([\d,.]+)\s*│\s*([\d,.]+)\s*│\s*([\d,.]+)\s*│' + ) + for line in result.stdout.split('\n'): + m = pattern.search(line) + if m: + ts, o, h, l, c, v = m.groups() + def parse_num(s): + return float(s.replace(',', '')) + klines.append({ + 'open': parse_num(o), + 'high': parse_num(h), + 'low': parse_num(l), + 'close': parse_num(c), + 'volume': parse_num(v), + }) + return klines + + +def fetch_klines_us(symbol: str, period: str = '5m', count: int = 30) -> list: + """美股 JSON""" + result = subprocess.run( + ['proxychains4', '-f', '/home/openclaw/.proxychains/proxychains.conf', + '/home/openclaw/.local/bin/longbridge', '--profile', 'lb_real', + 'candlesticks', symbol, period, '--count', str(count), '--json'], + capture_output=True, text=True, timeout=30, + ) + start = result.stdout.find('[') + if start == -1: + return [] + try: + data = json.loads(result.stdout[start:]) + return [{ + 'open': float(k['open']), + 'high': float(k['high']), + 'low': float(k['low']), + 'close': float(k['close']), + 'volume': float(k.get('volume', 0)), + } for k in data if 'close' in k] + except Exception: + return [] + + +def fetch_quote(symbol: str) -> dict: + result = subprocess.run( + ['proxychains4', '-f', '/home/openclaw/.proxychains/proxychains.conf', + '/home/openclaw/.local/bin/longbridge', '--profile', 'lb_real', + 'quote', symbol, '--json'], + capture_output=True, text=True, timeout=30, + ) + text = result.stdout + start = text.find('[') + if start == -1: + return {} + try: + return json.loads(text[start:])[0] + except Exception: + return {} + + +def analyze_market(symbol: str, market: str, klines: list, quote: dict) -> str: + """返回单支票分析报告""" + if not klines or not quote: + return f"❌ {symbol} 数据缺失" + + try: + import pandas as pd + df = pd.DataFrame(klines) + except ImportError: + return f"❌ pandas 未装" + + prev_close = quote.get('prev_close', 0) + current_price = quote['last_done'] + open_p = quote['open'] + gap_pct = ((open_p - prev_close) / prev_close * 100) if prev_close else 0 + + selector = IntradayStrategySelector() + diag = selector.diagnose(df, open_gap_pct=gap_pct) + + # 策略 emoji + strategy_emoji = { + StrategyType.TREND_FOLLOWING: '📈', + StrategyType.GRID_TRADING: '🔲', + StrategyType.BOLLINGER_REVERSION: '📊', + StrategyType.NO_TRADE: '⛔', + } + regime_short = { + MarketRegime.STRONG_TREND_UP: '强趋↑', + MarketRegime.STRONG_TREND_DOWN: '强趋↓', + MarketRegime.HIGH_VOL_SHAKE: '高波震荡', + MarketRegime.LOW_VOL_STABLE: '低波震荡', + MarketRegime.CHAOTIC: '混乱', + MarketRegime.UNKNOWN: '未知', + } + + params_str = '\n'.join(f" {k}: {v}" for k, v in diag.strategy_params.items()) + + return ( + f"\n{strategy_emoji.get(diag.recommended_strategy, '•')} **{symbol}** ({market}) " + f"现价 ${current_price:.2f} ({gap_pct:+.2f}%) " + f"置信度 {diag.confidence:.0%}\n" + f" 状态: {regime_short.get(diag.regime, diag.regime.value)} | " + f"R²={diag.r_squared} | 波动率={diag.volatility:.2%} | ADF p={diag.adf_pvalue}\n" + f" 推荐: {diag.recommended_strategy.value}\n" + f"{params_str}" + ) + + +def scan_market(market: str, candidate_file: Path, fetch_klines_func) -> list: + """扫描一个市场""" + if not candidate_file.exists(): + return [f"⚠️ 候选池不存在: {candidate_file.name}"] + + with open(candidate_file) as f: + data = json.load(f) + + results = data.get('results', [])[:5] # top 5 + date = data.get('date', '?')[:10] + + if not results: + return [f"⚠️ {market} 候选池为空"] + + reports = [f"📊 {market} 日内市场状态扫描 ({date})"] + + for entry in results: + symbol = entry['ticker'] + score = entry['score'] + try: + quote = fetch_quote(symbol) + klines = fetch_klines_func(symbol, '5m', 30) + report = analyze_market(symbol, market, klines, quote) + reports.append(report) + except Exception as e: + reports.append(f"❌ {symbol} 异常: {e}") + + return reports + + +def main(): + # 港股 + 美股 + hk_reports = scan_market('HK', CANDIDATE_HK, fetch_klines_hk) + us_reports = scan_market('US', CANDIDATE_US, fetch_klines_us) + + print(f"📊 日内市场状态扫描 ({hk_reports[0].split('(')[-1].rstrip(')')})\n") + print('=' * 60) + + print('\n--- 港股 ---') + for r in hk_reports[1:]: + print(r) + print() + + print('\n--- 美股 ---') + for r in us_reports[1:]: + print(r) + + +if __name__ == '__main__': + main() \ No newline at end of file diff --git a/strategy-management/scripts/exit_levels.py b/strategy-management/scripts/exit_levels.py new file mode 100644 index 0000000..611bbca --- /dev/null +++ b/strategy-management/scripts/exit_levels.py @@ -0,0 +1,292 @@ +""" +exit_levels.py - 港美股日内做T 出场点位计算 (混合公式) + +设计: 方法 2 (百分比波动率) + 方法 3 (关键价位) 混合 +- 关键位: day_high/low / prev_high/low / VWAP (从 indicators.py) +- 波动率: ATR (从 indicators.py) +- R:R 强制下限 1.5, 不达标信号否决 + +⚠️ 这是港美股做T专用, 币圈用 crypto-t-monitor 的 ATR 公式 (独立) + +用法: + from exit_levels import calc_exit_levels + + result = calc_exit_levels( + entry=100.0, + atr=5.0, + current_price=100.0, + day_high=103.0, + day_low=97.0, + prev_high=106.0, + prev_low=94.0, + vwap=101.0, + side='long', + min_rr=1.5, + ) + if result is None: + print("信号否决: R:R 不达标") + else: + sl, tp1, tp2 = result + print(f"SL={sl} TP1={tp1} TP2={tp2}") +""" + +from dataclasses import dataclass +from typing import Optional, Literal + + +@dataclass +class ExitLevels: + """出场点位结果""" + sl: float # 止损价位 + tp1: float # 第一止盈 (半平) + tp2: float # 第二止盈 (全平) + entry: float # 入场价 (回填, 方便调用方记录) + side: str # 'long' / 'short' + risk: float # 风险 (entry - SL) + reward: float # 奖励 (TP1 - entry) + rr_ratio: float # R:R (reward/risk) + sl_method: str # 'vol_pct' | 'key_level' | 'hybrid' + tp_method: str # 'vol_pct' | 'key_level' | 'hybrid' + note: str = "" # 备注 (VWAP 锁定等) + + +def calc_exit_levels( + entry: float, + atr: float, + current_price: float, + day_high: Optional[float] = None, + day_low: Optional[float] = None, + prev_high: Optional[float] = None, + prev_low: Optional[float] = None, + vwap: Optional[float] = None, + side: Literal['long', 'short'] = 'long', + min_rr: float = 1.5, + # 方法 2 权重 (百分比波动率) + vol_sl_multi: float = 1.0, # SL 距离 = vol × sl_multi + vol_tp1_multi: float = 2.0, # TP1 距离 = vol × tp1_multi (默认 2.0 倍 SL → R:R 2:1) + vol_tp2_multi: float = 3.0, # TP2 距离 = vol × tp2_multi + # 方法 3 权重 (关键位) - 离关键位的 buffer + key_level_buffer_pct: float = 0.001, # 0.1% 缓冲 (避免瞬时触发) +) -> Optional[ExitLevels]: + """ + 计算 SL/TP1/TP2 (混合公式: 波动率 + 关键位) + + Args: + entry: 入场价 (假设已知, 或用 adjust_to_ask1 拿到的价) + atr: 当前 K 线 ATR (14 周期, 从 indicators.atr()) + current_price: 当前实时价 (用于 vol_pct 计算) + day_high/low: 今日最高/最低 (从 longbridge quote) + prev_high/low: 昨日最高/最低 (从 longbridge K 线) + vwap: 成交量加权平均价 (从 indicators.vwap()) + side: 'long' (做多) / 'short' (做空) + min_rr: 最小 R:R (默认 1.5) + vol_sl_multi / tp1_multi / tp2_multi: ATR 倍数 + key_level_buffer_pct: 关键位 buffer (避免价格精确等于关键位) + + Returns: + ExitLevels 或 None (R:R 不达标) + + Examples: + >>> calc_exit_levels(entry=100, atr=5, current_price=100, + ... day_high=115, day_low=92, prev_high=118, prev_low=90, + ... vwap=105, side='long') + ExitLevels(sl=91.2, tp1=114.2, tp2=120.0, ...) + """ + if entry <= 0 or atr <= 0 or current_price <= 0: + raise ValueError("entry/atr/current_price must be > 0") + + # === 方法 2: 百分比波动率 (基于 ATR) === + vol_pct = atr / current_price + vol_sl_dist = vol_pct * vol_sl_multi + vol_tp1_dist = vol_pct * vol_tp1_multi + vol_tp2_dist = vol_pct * vol_tp2_multi + + if side == 'long': + sl_vol = entry * (1 - vol_sl_dist) + tp1_vol = entry * (1 + vol_tp1_dist) + tp2_vol = entry * (1 + vol_tp2_dist) + else: + sl_vol = entry * (1 + vol_sl_dist) + tp1_vol = entry * (1 - vol_tp1_dist) + tp2_vol = entry * (1 - vol_tp2_dist) + + # === 方法 3: 关键价位 === + # 多仓: SL 取 entry **下方**的支撑; TP1 取 entry **上方**的阻力 + # 空仓: SL 取 entry **上方**的阻力; TP1 取 entry **下方**的支撑 + sl_key = None + tp1_key = None + note = "" + + if side == 'long': + # SL 关键位 (entry 下方) + sl_candidates = [] + if day_low is not None and day_low < entry: + sl_candidates.append(day_low * (1 - key_level_buffer_pct)) + if prev_low is not None and prev_low < entry: + sl_candidates.append(prev_low * (1 - key_level_buffer_pct)) + if vwap is not None and vwap < entry: + sl_candidates.append(vwap * (1 - key_level_buffer_pct)) + sl_key = max(sl_candidates) if sl_candidates else None + + # TP1 关键位 (entry 上方) + tp1_candidates = [] + if day_high is not None and day_high > entry: + tp1_candidates.append(day_high * (1 - key_level_buffer_pct)) + if prev_high is not None and prev_high > entry: + tp1_candidates.append(prev_high * (1 - key_level_buffer_pct)) + if vwap is not None and vwap > entry: + tp1_candidates.append(vwap * (1 - key_level_buffer_pct)) + tp1_key = min(tp1_candidates) if tp1_candidates else None + else: # short + # SL 关键位 (entry 上方, 空仓止损 = 价格涨到这里平) + sl_candidates = [] + if day_high is not None and day_high > entry: + sl_candidates.append(day_high * (1 + key_level_buffer_pct)) + if prev_high is not None and prev_high > entry: + sl_candidates.append(prev_high * (1 + key_level_buffer_pct)) + if vwap is not None and vwap > entry: + sl_candidates.append(vwap * (1 + key_level_buffer_pct)) + sl_key = min(sl_candidates) if sl_candidates else None + + # TP1 关键位 (entry 下方) + tp1_candidates = [] + if day_low is not None and day_low < entry: + tp1_candidates.append(day_low * (1 + key_level_buffer_pct)) + if prev_low is not None and prev_low < entry: + tp1_candidates.append(prev_low * (1 + key_level_buffer_pct)) + if vwap is not None and vwap < entry: + tp1_candidates.append(vwap * (1 + key_level_buffer_pct)) + tp1_key = max(tp1_candidates) if tp1_candidates else None + + # === 混合: vol_pct 主导 (70%), 关键位微调 (30%) === + # 关键位 30% 权重, 防止 VWAP 等动态位锁死 + # vol_pct 至少占 70% (最终值不会偏离 vol_pct 太远) + if side == 'long': + if sl_key is not None: + SL = sl_vol * 0.7 + sl_key * 0.3 + sl_method = 'hybrid_blend' + else: + SL = sl_vol + sl_method = 'vol_pct' + + if tp1_key is not None: + TP1 = tp1_vol * 0.7 + tp1_key * 0.3 + tp_method = 'hybrid_blend' + else: + TP1 = tp1_vol + tp_method = 'vol_pct' + + TP2 = tp2_vol + else: # short + if sl_key is not None: + SL = sl_vol * 0.7 + sl_key * 0.3 + sl_method = 'hybrid_blend' + else: + SL = sl_vol + sl_method = 'vol_pct' + + if tp1_key is not None: + TP1 = tp1_vol * 0.7 + tp1_key * 0.3 + tp_method = 'hybrid_blend' + else: + TP1 = tp1_vol + tp_method = 'vol_pct' + + TP2 = tp2_vol + + # === R:R 检查 === + if side == 'long': + risk = entry - SL + reward = TP1 - entry + else: + risk = SL - entry + reward = entry - TP1 + + if risk <= 0: + return None # 止损 >= 入场 (逻辑错误) + + rr_ratio = reward / risk if risk > 0 else 0 + + # 风险 vs reward + if abs(reward) < min_rr * abs(risk): + # R:R 不达标 + if sl_key == tp1_key and sl_key is not None: + note = f"VWAP 既作支撑又作阻力, 价格窄幅震荡 (SL=TP1={sl_key:.2f})" + else: + note = f"R:R {rr_ratio:.2f} < {min_rr}, 信号否决" + return None + + return ExitLevels( + sl=SL, + tp1=TP1, + tp2=TP2, + entry=entry, + side=side, + risk=abs(risk), + reward=abs(reward), + rr_ratio=rr_ratio, + sl_method=sl_method, + tp_method=tp_method, + note=note, + ) + + +def format_levels(levels: ExitLevels) -> str: + """格式化输出 (QQ 推送用)""" + side_emoji = '🟢' if levels.side == 'long' else '🔴' + return ( + f"{side_emoji} {levels.side.upper()} @ ${levels.entry:.2f}\n" + f" SL: ${levels.sl:.2f} ({levels.sl_method})\n" + f" TP1: ${levels.tp1:.2f} ({levels.tp_method})\n" + f" TP2: ${levels.tp2:.2f}\n" + f" Risk/Reward: 1:{levels.rr_ratio:.2f}" + ) + + +if __name__ == '__main__': + # 自检: 用 indicators.py 的真实数据测试 + print("=" * 60) + print("exit_levels.py - 自检") + print("=" * 60) + + # 案例 1: NVDA 高波动 (有 R:R) + print("\n[案例 1] NVDA $100, ATR=$8, day H/L=$115/$92, prev H/L=$118/$90") + result = calc_exit_levels( + entry=100.0, atr=8.0, current_price=100.0, + day_high=115.0, day_low=92.0, + prev_high=118.0, prev_low=90.0, + vwap=105.0, + side='long', + ) + if result: + print(format_levels(result)) + else: + print("❌ 信号否决") + + # 案例 2: 价在 VWAP 上下窄幅震荡 + print("\n[案例 2] NVDA $100, ATR=$3, VWAP=$101 (紧贴)") + result = calc_exit_levels( + entry=100.0, atr=3.0, current_price=100.0, + day_high=103.0, day_low=98.0, + prev_high=105.0, prev_low=95.0, + vwap=101.0, + side='long', + ) + if result: + print(format_levels(result)) + else: + print("❌ 信号否决 (R:R 不达标 / VWAP 锁定)") + + # 案例 3: 空仓 + 大波动 + print("\n[案例 3] TSDA short $200, ATR=$12") + result = calc_exit_levels( + entry=200.0, atr=12.0, current_price=200.0, + day_high=212.0, day_low=188.0, + prev_high=215.0, prev_low=185.0, + vwap=205.0, + side='short', + ) + if result: + print(format_levels(result)) + else: + print("❌ 信号否决") \ No newline at end of file