303 lines
10 KiB
Python
303 lines
10 KiB
Python
"""Relative Strength vs BTC — 相对强度因子。
|
||
|
||
核心逻辑：
|
||
- 计算币种过去 N 天涨幅 vs BTC 同期涨幅的差值
|
||
- 标准化为 RS 分数（正值=强于BTC，负值=弱于BTC）
|
||
- 提供 RS 排名百分位，用于筛选前置条件
|
||
|
||
使用场景：
|
||
- RS 排名前 30% 的币 → 加分（潜在领涨）
|
||
- RS 排名后 50% 的币 → 降级（弱势币即使出信号也成功率低）
|
||
- BTC 回调时 RS 仍为正 → 独立买盘，高价值信号
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import time
|
||
from typing import Optional
|
||
|
||
import pandas as pd
|
||
import requests
|
||
|
||
from app.config.config_loader import _get_section as _get_cfg_section
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Configuration
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _rs_config() -> dict:
|
||
"""Load RS config from rules.yaml -> relative_strength section."""
|
||
try:
|
||
cfg = _get_cfg_section("relative_strength") or {}
|
||
except Exception:
|
||
cfg = {}
|
||
return {
|
||
"lookback_days": int(cfg.get("lookback_days", 5)),
|
||
"short_lookback_days": int(cfg.get("short_lookback_days", 3)),
|
||
"strong_threshold": float(cfg.get("strong_threshold", 5.0)),
|
||
"weak_threshold": float(cfg.get("weak_threshold", -3.0)),
|
||
"top_percentile": float(cfg.get("top_percentile", 30)),
|
||
"bottom_percentile": float(cfg.get("bottom_percentile", 50)),
|
||
"weight_strong": float(cfg.get("weight_strong", 3.0)),
|
||
"weight_weak_penalty": float(cfg.get("weight_weak_penalty", -2.0)),
|
||
"btc_drawdown_bonus": float(cfg.get("btc_drawdown_bonus", 2.0)),
|
||
"btc_drawdown_threshold": float(cfg.get("btc_drawdown_threshold", -2.0)),
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Data fetching
|
||
# ---------------------------------------------------------------------------
|
||
|
||
BINANCE_SPOT_BASE = "https://api.binance.com"
|
||
|
||
# Simple in-memory cache for BTC klines (refreshed every 5 min)
|
||
_btc_cache: dict = {"data": None, "ts": 0}
|
||
|
||
|
||
def _fetch_daily_klines(symbol_raw: str, limit: int = 10) -> Optional[pd.DataFrame]:
|
||
"""Fetch daily klines from Binance spot public API.
|
||
|
||
Args:
|
||
symbol_raw: Binance raw symbol like "BTCUSDT"
|
||
limit: number of daily candles
|
||
"""
|
||
try:
|
||
url = f"{BINANCE_SPOT_BASE}/api/v3/klines"
|
||
resp = requests.get(url, params={
|
||
"symbol": symbol_raw,
|
||
"interval": "1d",
|
||
"limit": limit,
|
||
}, timeout=8)
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
if not data:
|
||
return None
|
||
df = pd.DataFrame(data, columns=[
|
||
"open_time", "open", "high", "low", "close", "volume",
|
||
"close_time", "quote_volume", "trades", "taker_buy_base",
|
||
"taker_buy_quote", "ignore",
|
||
])
|
||
df["open"] = df["open"].astype(float)
|
||
df["close"] = df["close"].astype(float)
|
||
df["high"] = df["high"].astype(float)
|
||
df["low"] = df["low"].astype(float)
|
||
return df
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _get_btc_daily(limit: int = 10) -> Optional[pd.DataFrame]:
|
||
"""Get BTC daily klines with simple caching (5 min TTL)."""
|
||
now = time.time()
|
||
if _btc_cache["data"] is not None and now - _btc_cache["ts"] < 300:
|
||
return _btc_cache["data"]
|
||
df = _fetch_daily_klines("BTCUSDT", limit=limit)
|
||
if df is not None:
|
||
_btc_cache["data"] = df
|
||
_btc_cache["ts"] = now
|
||
return df
|
||
|
||
|
||
def _change_over_n_days(df: pd.DataFrame, n: int) -> Optional[float]:
|
||
"""Calculate percentage change over last N completed daily candles.
|
||
|
||
Uses close of (today - N) vs close of yesterday (last completed candle).
|
||
The last row might be the current incomplete day, so we use iloc[-2] as
|
||
the latest completed close and iloc[-(n+1)] as the start.
|
||
"""
|
||
if df is None or len(df) < n + 1:
|
||
return None
|
||
try:
|
||
# iloc[-1] = current (possibly incomplete) day
|
||
# iloc[-2] = last completed day
|
||
# For N-day change: compare close at -(N+1) to close at -1
|
||
start_close = float(df["close"].iloc[-(n + 1)])
|
||
end_close = float(df["close"].iloc[-1])
|
||
if start_close <= 0:
|
||
return None
|
||
return (end_close - start_close) / start_close * 100
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Core RS calculation
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def compute_relative_strength(
|
||
symbol: str,
|
||
coin_df_daily: Optional[pd.DataFrame] = None,
|
||
btc_df_daily: Optional[pd.DataFrame] = None,
|
||
) -> dict:
|
||
"""Compute RS metrics for a single coin vs BTC.
|
||
|
||
Args:
|
||
symbol: e.g. "SOL/USDT"
|
||
coin_df_daily: pre-fetched daily klines for the coin (optional)
|
||
btc_df_daily: pre-fetched BTC daily klines (optional)
|
||
|
||
Returns:
|
||
{
|
||
"rs_5d": float, # 5-day RS (coin_change - btc_change)
|
||
"rs_3d": float, # 3-day RS
|
||
"rs_score": float, # weighted composite RS
|
||
"btc_change_5d": float,
|
||
"coin_change_5d": float,
|
||
"btc_drawdown": bool, # True if BTC is down significantly
|
||
"independent_strength": bool, # coin up while BTC down
|
||
}
|
||
"""
|
||
cfg = _rs_config()
|
||
lookback = cfg["lookback_days"]
|
||
short_lookback = cfg["short_lookback_days"]
|
||
|
||
# Fetch BTC data if not provided
|
||
if btc_df_daily is None:
|
||
btc_df_daily = _get_btc_daily(limit=lookback + 2)
|
||
|
||
# Fetch coin data if not provided
|
||
if coin_df_daily is None:
|
||
raw_symbol = symbol.replace("/", "")
|
||
coin_df_daily = _fetch_daily_klines(raw_symbol, limit=lookback + 2)
|
||
|
||
# Calculate changes
|
||
btc_change_long = _change_over_n_days(btc_df_daily, lookback)
|
||
btc_change_short = _change_over_n_days(btc_df_daily, short_lookback)
|
||
coin_change_long = _change_over_n_days(coin_df_daily, lookback)
|
||
coin_change_short = _change_over_n_days(coin_df_daily, short_lookback)
|
||
|
||
if btc_change_long is None or coin_change_long is None:
|
||
return {
|
||
"rs_5d": 0.0,
|
||
"rs_3d": 0.0,
|
||
"rs_score": 0.0,
|
||
"btc_change_5d": 0.0,
|
||
"coin_change_5d": 0.0,
|
||
"btc_drawdown": False,
|
||
"independent_strength": False,
|
||
"available": False,
|
||
}
|
||
|
||
rs_long = round(coin_change_long - btc_change_long, 2)
|
||
rs_short = round((coin_change_short or 0) - (btc_change_short or 0), 2)
|
||
|
||
# Composite: weight short-term RS slightly more for 1-3 day trades
|
||
rs_score = round(rs_short * 0.6 + rs_long * 0.4, 2)
|
||
|
||
# BTC drawdown detection
|
||
btc_drawdown = (btc_change_short or 0) < cfg["btc_drawdown_threshold"]
|
||
|
||
# Independent strength: coin is positive while BTC is negative
|
||
independent_strength = (
|
||
btc_drawdown
|
||
and (coin_change_short or 0) > 0
|
||
)
|
||
|
||
return {
|
||
"rs_5d": rs_long,
|
||
"rs_3d": rs_short,
|
||
"rs_score": rs_score,
|
||
"btc_change_5d": round(btc_change_long, 2),
|
||
"btc_change_3d": round(btc_change_short or 0, 2),
|
||
"coin_change_5d": round(coin_change_long, 2),
|
||
"coin_change_3d": round(coin_change_short or 0, 2),
|
||
"btc_drawdown": btc_drawdown,
|
||
"independent_strength": independent_strength,
|
||
"available": True,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Batch RS ranking
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def rank_universe_rs(
|
||
symbols: list[str],
|
||
btc_df_daily: Optional[pd.DataFrame] = None,
|
||
) -> dict[str, dict]:
|
||
"""Compute RS for a list of symbols and attach percentile rank.
|
||
|
||
Returns: {symbol: {rs_score, rs_percentile, rs_tier, ...}}
|
||
where rs_tier is "strong" / "neutral" / "weak".
|
||
"""
|
||
cfg = _rs_config()
|
||
|
||
if btc_df_daily is None:
|
||
btc_df_daily = _get_btc_daily(limit=cfg["lookback_days"] + 2)
|
||
|
||
results = {}
|
||
for symbol in symbols:
|
||
rs = compute_relative_strength(symbol, btc_df_daily=btc_df_daily)
|
||
results[symbol] = rs
|
||
|
||
# Calculate percentile ranks
|
||
scores = [r["rs_score"] for r in results.values() if r.get("available")]
|
||
if not scores:
|
||
return results
|
||
|
||
sorted_scores = sorted(scores)
|
||
n = len(sorted_scores)
|
||
|
||
for symbol, rs_data in results.items():
|
||
if not rs_data.get("available"):
|
||
rs_data["rs_percentile"] = 50.0
|
||
rs_data["rs_tier"] = "neutral"
|
||
continue
|
||
|
||
score = rs_data["rs_score"]
|
||
# Percentile: what % of coins have lower RS
|
||
rank = sum(1 for s in sorted_scores if s < score)
|
||
percentile = round(rank / n * 100, 1) if n > 0 else 50.0
|
||
rs_data["rs_percentile"] = percentile
|
||
|
||
if percentile >= (100 - cfg["top_percentile"]):
|
||
rs_data["rs_tier"] = "strong"
|
||
elif percentile <= cfg["bottom_percentile"]:
|
||
rs_data["rs_tier"] = "weak"
|
||
else:
|
||
rs_data["rs_tier"] = "neutral"
|
||
|
||
return results
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Factor scoring interface
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def rs_factor_score(rs_data: dict) -> tuple[float, str]:
|
||
"""Convert RS data into a factor score for the scoring system.
|
||
|
||
Returns:
|
||
(score_delta, signal_label)
|
||
- Positive score for strong RS
|
||
- Negative score for weak RS (penalty)
|
||
- Extra bonus if independent strength during BTC drawdown
|
||
"""
|
||
cfg = _rs_config()
|
||
|
||
if not rs_data.get("available"):
|
||
return 0.0, ""
|
||
|
||
score = 0.0
|
||
labels = []
|
||
|
||
tier = rs_data.get("rs_tier", "neutral")
|
||
rs_val = rs_data.get("rs_score", 0)
|
||
|
||
if tier == "strong":
|
||
score += cfg["weight_strong"]
|
||
labels.append(f"RS强势({rs_val:+.1f}%)")
|
||
elif tier == "weak":
|
||
score += cfg["weight_weak_penalty"]
|
||
labels.append(f"RS弱势({rs_val:+.1f}%)")
|
||
|
||
# Bonus: coin holds up while BTC drops
|
||
if rs_data.get("independent_strength"):
|
||
score += cfg["btc_drawdown_bonus"]
|
||
labels.append("BTC回调中独立走强")
|
||
|
||
label = " | ".join(labels) if labels else ""
|
||
return round(score, 2), label
|