dekun/crypto_monitor
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add long-straddle premium overlay to hub amp stats.

Browse Source 
Configurable bilateral premium with exceed counts/ratios and settlement PnL for buying volatility.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-07-23 02:33:23 +08:00
parent 61e8da1e8b
commit 789ab43dbe
7 changed files with 347 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/hub/amp_stats_lib.py
+101 -3
View File
@@ -165,9 +165,74 @@ def compute_day_row(
}
def summarize_rows(rows: list[dict[str, Any]]) -> dict[str, Any]:
def normalize_straddle_premium(raw: Any) -> Optional[float]:
"""双边权利金(点数).空/≤0 表示不做跨式对照."""
if raw is None or raw == "":
return None
try:
v = float(raw)
except (TypeError, ValueError):
raise ValueError("双边权利金须为数字") from None
if v <= 0:
return None
return v
def straddle_long_stats(rows: list[dict[str, Any]], premium: float) -> dict[str, Any]:
"""买跨(赌波动):盈亏按收盘 |C−O| − 双边权利金;越过阈值用严格 >."""
prem = float(premium)
if prem <= 0:
raise ValueError("双边权利金须 > 0")
if not rows:
return {
"side": "long_straddle",
"premium": prem,
"sample_count": 0,
"up_exceed_days": 0,
"up_exceed_ratio": None,
"down_exceed_days": 0,
"down_exceed_ratio": None,
"abs_change_exceed_days": 0,
"abs_change_exceed_ratio": None,
"pnl_total": None,
"pnl_avg": None,
"win_days": 0,
"win_ratio": None,
"pnl_max": None,
"pnl_min": None,
}
n = len(rows)
up_ex = sum(1 for r in rows if float(r["up_points"]) > prem)
down_ex = sum(1 for r in rows if float(r["down_points"]) > prem)
abs_ex = sum(1 for r in rows if abs(float(r["change"])) > prem)
pnls = [abs(float(r["change"])) - prem for r in rows]
win = sum(1 for p in pnls if p > 0)
return {
"side": "long_straddle",
"premium": round(prem, 4),
"sample_count": n,
"up_exceed_days": up_ex,
"up_exceed_ratio": round(up_ex / n, 4),
"down_exceed_days": down_ex,
"down_exceed_ratio": round(down_ex / n, 4),
"abs_change_exceed_days": abs_ex,
"abs_change_exceed_ratio": round(abs_ex / n, 4),
"pnl_total": round(sum(pnls), 4),
"pnl_avg": round(statistics.fmean(pnls), 4),
"win_days": win,
"win_ratio": round(win / n, 4),
"pnl_max": round(max(pnls), 4),
"pnl_min": round(min(pnls), 4),
}
def summarize_rows(
rows: list[dict[str, Any]],
*,
straddle_premium: Any = None,
) -> dict[str, Any]:
if not rows:
out = {
"sample_count": 0,
"max_amplitude": None,
"max_amplitude_day": None,
@@ -179,7 +244,12 @@ def summarize_rows(rows: list[dict[str, Any]]) -> dict[str, Any]:
"avg_down_points": None,
"up_day_ratio": None,
"down_day_ratio": None,
"straddle": None,
}
prem = normalize_straddle_premium(straddle_premium)
if prem is not None:
out["straddle"] = straddle_long_stats([], prem)
return out
amps = [float(r["amplitude"]) for r in rows]
ups = [float(r["up_points"]) for r in rows]
downs = [float(r["down_points"]) for r in rows]
@@ -188,7 +258,7 @@ def summarize_rows(rows: list[dict[str, Any]]) -> dict[str, Any]:
up_days = sum(1 for r in rows if float(r["change"]) > 0)
down_days = sum(1 for r in rows if float(r["change"]) < 0)
n = len(rows)
return {
out: dict[str, Any] = {
"sample_count": n,
"max_amplitude": round(max_amp, 4),
"max_amplitude_day": max_amp_day,
@@ -200,7 +270,12 @@ def summarize_rows(rows: list[dict[str, Any]]) -> dict[str, Any]:
"avg_down_points": round(statistics.fmean(downs), 4),
"up_day_ratio": round(up_days / n, 4),
"down_day_ratio": round(down_days / n, 4),
"straddle": None,
}
prem = normalize_straddle_premium(straddle_premium)
if prem is not None:
out["straddle"] = straddle_long_stats(rows, prem)
return out
def _parse_okx_candle_row(row: list) -> Optional[dict[str, Any]]:
@@ -312,6 +387,7 @@ def compute_amp_stats(
start_hour: int = 16,
period: str = "2m",
custom_days: Any = None,
straddle_premium: Any = None,
now: Optional[datetime] = None,
fetch_fn: Optional[Callable[..., list[dict[str, Any]]]] = None,
) -> dict[str, Any]:
@@ -319,6 +395,7 @@ def compute_amp_stats(
sh = int(start_hour)
if sh < 0 or sh > 23:
raise ValueError("起点须为 0-23 整点")
prem = normalize_straddle_premium(straddle_premium)
sample_days = resolve_sample_days(period, custom_days)
settlements = list_settlement_dates(sample_days=sample_days, now=now)
if not settlements:
@@ -342,7 +419,7 @@ def compute_amp_stats(
missing.append(d.isoformat())
continue
rows.append(row)
summary = summarize_rows(rows)
summary = summarize_rows(rows, straddle_premium=prem)
period_label = period if period != "custom" else f"custom:{sample_days}"
return {
"ok": True,
@@ -353,6 +430,7 @@ def compute_amp_stats(
"end_hour": END_HOUR,
"period": period_label,
"sample_days_requested": sample_days,
"straddle_premium": prem,
"timeframe": TIMEFRAME,
"price_source": price_source,
"inst_id": inst_id,
@@ -398,6 +476,26 @@ def build_export_csv(payload: dict[str, Any]) -> str:
w.writerow(["开→高最大", s.get("max_up_points"), "均值", s.get("avg_up_points")])
w.writerow(["开→低最大", s.get("max_down_points"), "均值", s.get("avg_down_points")])
w.writerow(["上涨窗占比", s.get("up_day_ratio"), "下跌窗占比", s.get("down_day_ratio")])
st = s.get("straddle") or {}
if st:
w.writerow([])
w.writerow(["【买跨对照·双边权利金】", st.get("premium")])
w.writerow(["开→高超过", st.get("up_exceed_days"), "占比", st.get("up_exceed_ratio")])
w.writerow(["开→低超过", st.get("down_exceed_days"), "占比", st.get("down_exceed_ratio")])
w.writerow(["|涨跌|超过", st.get("abs_change_exceed_days"), "占比", st.get("abs_change_exceed_ratio")])
w.writerow(
[
"买跨点数盈亏合计",
st.get("pnl_total"),
"日均",
st.get("pnl_avg"),
"赚钱天数",
st.get("win_days"),
"胜率",
st.get("win_ratio"),
]
)
w.writerow(["单日最大赚", st.get("pnl_max"), "单日最大亏", st.get("pnl_min")])
w.writerow([])
w.writerow(["【日表明细】"])
w.writerow(