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

Add weekend filter, take-profit, and profit column to amp stats.

Browse Source 
Long-straddle effective move uses TP on path hit (>=) else abs change; mark Sat/Sun on settlement days.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-07-23 02:47:20 +08:00
parent 789ab43dbe
commit b64c742fc9
7 changed files with 512 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/hub/amp_stats_lib.py
+222 -23
View File
@@ -150,10 +150,15 @@ def compute_day_row(
down = opens - lo
amp = hi - lo
change = close - opens
wd = settlement.weekday() # Mon=0 … Sun=6
is_we = wd >= 5
return {
"settlement_day": settlement.isoformat(),
"window_start": start.strftime("%Y-%m-%d %H:%M"),
"window_end": end.strftime("%Y-%m-%d %H:%M"),
"weekday": wd,
"weekday_label": "六" if wd == 5 else ("日" if wd == 6 else ""),
"is_weekend": is_we,
"open": round(opens, 4),
"high": round(hi, 4),
"low": round(lo, 4),
@@ -178,15 +183,113 @@ def normalize_straddle_premium(raw: Any) -> Optional[float]:
return v
def straddle_long_stats(rows: list[dict[str, Any]], premium: float) -> dict[str, Any]:
"""买跨(赌波动):盈亏按收盘 |C−O| − 双边权利金;越过阈值用严格 >."""
def normalize_take_profit(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 normalize_weekend_filter(raw: Any) -> str:
"""all | exclude | only;默认全部."""
s = (str(raw) if raw is not None else "all").strip().lower()
if s in ("", "all", "全部"):
return "all"
if s in ("exclude", "exclude_weekend", "no_weekend", "排除周末"):
return "exclude"
if s in ("only", "weekend_only", "only_weekend", "仅周末"):
return "only"
raise ValueError("周末筛选须为 all / exclude / only")
def filter_weekend_rows(rows: list[dict[str, Any]], weekend_filter: Any = "all") -> list[dict[str, Any]]:
mode = normalize_weekend_filter(weekend_filter)
if mode == "all":
return list(rows or [])
out: list[dict[str, Any]] = []
for r in rows or []:
is_we = bool(r.get("is_weekend"))
if "is_weekend" not in r and r.get("settlement_day"):
try:
is_we = date.fromisoformat(str(r["settlement_day"])).weekday() >= 5
except ValueError:
is_we = False
if mode == "exclude" and is_we:
continue
if mode == "only" and not is_we:
continue
out.append(r)
return out
def effective_move_points(row: dict[str, Any], take_profit: Optional[float]) -> float:
"""触达止盈(≥)用止盈点,否则用 |涨跌|."""
abs_chg = abs(float(row.get("change") or 0))
if take_profit is None:
return abs_chg
tp = float(take_profit)
up = float(row.get("up_points") or 0)
down = float(row.get("down_points") or 0)
if up >= tp or down >= tp:
return tp
return abs_chg
def enrich_rows_pnl(
rows: list[dict[str, Any]],
*,
straddle_premium: Optional[float] = None,
take_profit: Optional[float] = None,
) -> list[dict[str, Any]]:
"""为日表附加有效波动 / 是否触达止盈 / 收益(有权利金时)."""
prem = normalize_straddle_premium(straddle_premium)
tp = normalize_take_profit(take_profit)
out: list[dict[str, Any]] = []
for r in rows or []:
item = dict(r)
if "is_weekend" not in item and item.get("settlement_day"):
try:
wd = date.fromisoformat(str(item["settlement_day"])).weekday()
item["weekday"] = wd
item["weekday_label"] = "六" if wd == 5 else ("日" if wd == 6 else "")
item["is_weekend"] = wd >= 5
except ValueError:
item.setdefault("weekday_label", "")
item.setdefault("is_weekend", False)
move = effective_move_points(item, tp)
hit = False
if tp is not None:
hit = float(item.get("up_points") or 0) >= tp or float(item.get("down_points") or 0) >= tp
item["effective_move"] = round(move, 4)
item["take_profit_hit"] = hit
item["profit"] = round(move - prem, 4) if prem is not None else None
out.append(item)
return out
def straddle_long_stats(
rows: list[dict[str, Any]],
premium: float,
*,
take_profit: Any = None,
) -> dict[str, Any]:
"""买跨:越过权利金用严格 >;收益=有效波动−权利金(止盈≥触达用止盈点,否则|涨跌|)."""
prem = float(premium)
if prem <= 0:
raise ValueError("双边权利金须 > 0")
if not rows:
tp = normalize_take_profit(take_profit)
enriched = enrich_rows_pnl(rows, straddle_premium=prem, take_profit=tp)
if not enriched:
return {
"side": "long_straddle",
"premium": prem,
"take_profit": tp,
"sample_count": 0,
"up_exceed_days": 0,
"up_exceed_ratio": None,
@@ -194,6 +297,8 @@ def straddle_long_stats(rows: list[dict[str, Any]], premium: float) -> dict[str,
"down_exceed_ratio": None,
"abs_change_exceed_days": 0,
"abs_change_exceed_ratio": None,
"tp_hit_days": 0,
"tp_hit_ratio": None,
"pnl_total": None,
"pnl_avg": None,
"win_days": 0,
@@ -201,15 +306,17 @@ def straddle_long_stats(rows: list[dict[str, Any]], premium: float) -> dict[str,
"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]
n = len(enriched)
up_ex = sum(1 for r in enriched if float(r["up_points"]) > prem)
down_ex = sum(1 for r in enriched if float(r["down_points"]) > prem)
abs_ex = sum(1 for r in enriched if abs(float(r["change"])) > prem)
tp_hits = sum(1 for r in enriched if r.get("take_profit_hit"))
pnls = [float(r["profit"]) for r in enriched if r.get("profit") is not None]
win = sum(1 for p in pnls if p > 0)
return {
"side": "long_straddle",
"premium": round(prem, 4),
"take_profit": round(tp, 4) if tp is not None else None,
"sample_count": n,
"up_exceed_days": up_ex,
"up_exceed_ratio": round(up_ex / n, 4),
@@ -217,6 +324,8 @@ def straddle_long_stats(rows: list[dict[str, Any]], premium: float) -> dict[str,
"down_exceed_ratio": round(down_ex / n, 4),
"abs_change_exceed_days": abs_ex,
"abs_change_exceed_ratio": round(abs_ex / n, 4),
"tp_hit_days": tp_hits,
"tp_hit_ratio": round(tp_hits / n, 4) if tp is not None else None,
"pnl_total": round(sum(pnls), 4),
"pnl_avg": round(statistics.fmean(pnls), 4),
"win_days": win,
@@ -230,6 +339,7 @@ def summarize_rows(
rows: list[dict[str, Any]],
*,
straddle_premium: Any = None,
take_profit: Any = None,
) -> dict[str, Any]:
if not rows:
out = {
@@ -248,7 +358,7 @@ def summarize_rows(
}
prem = normalize_straddle_premium(straddle_premium)
if prem is not None:
out["straddle"] = straddle_long_stats([], prem)
out["straddle"] = straddle_long_stats([], prem, take_profit=take_profit)
return out
amps = [float(r["amplitude"]) for r in rows]
ups = [float(r["up_points"]) for r in rows]
@@ -274,7 +384,7 @@ def summarize_rows(
}
prem = normalize_straddle_premium(straddle_premium)
if prem is not None:
out["straddle"] = straddle_long_stats(rows, prem)
out["straddle"] = straddle_long_stats(rows, prem, take_profit=take_profit)
return out
@@ -388,6 +498,8 @@ def compute_amp_stats(
period: str = "2m",
custom_days: Any = None,
straddle_premium: Any = None,
take_profit: Any = None,
weekend_filter: Any = "all",
now: Optional[datetime] = None,
fetch_fn: Optional[Callable[..., list[dict[str, Any]]]] = None,
) -> dict[str, Any]:
@@ -396,6 +508,8 @@ def compute_amp_stats(
if sh < 0 or sh > 23:
raise ValueError("起点须为 0-23 整点")
prem = normalize_straddle_premium(straddle_premium)
tp = normalize_take_profit(take_profit)
we_mode = normalize_weekend_filter(weekend_filter)
sample_days = resolve_sample_days(period, custom_days)
settlements = list_settlement_dates(sample_days=sample_days, now=now)
if not settlements:
@@ -411,37 +525,109 @@ def compute_amp_stats(
key, since_ms=since_ms, until_ms=until_ms, fetch_fn=fetch_fn
)
bar_map = bars_to_map(bars)
rows: list[dict[str, Any]] = []
rows_all: list[dict[str, Any]] = []
missing: list[str] = []
for d in settlements:
row = compute_day_row(d, sh, bar_map)
if row is None:
missing.append(d.isoformat())
continue
rows.append(row)
summary = summarize_rows(rows, straddle_premium=prem)
period_label = period if period != "custom" else f"custom:{sample_days}"
rows_all.append(row)
return build_amp_result(
rows_all=rows_all,
symbol_key=key,
start_hour=sh,
period=period,
sample_days=sample_days,
straddle_premium=prem,
take_profit=tp,
weekend_filter=we_mode,
price_source=price_source,
inst_id=inst_id,
missing=missing,
)
def build_amp_result(
*,
rows_all: list[dict[str, Any]],
symbol_key: str,
start_hour: int,
period: str,
sample_days: int,
straddle_premium: Any = None,
take_profit: Any = None,
weekend_filter: Any = "all",
price_source: str = "",
inst_id: str = "",
missing: Optional[list[str]] = None,
) -> dict[str, Any]:
prem = normalize_straddle_premium(straddle_premium)
tp = normalize_take_profit(take_profit)
we_mode = normalize_weekend_filter(weekend_filter)
filtered = filter_weekend_rows(rows_all, we_mode)
rows = enrich_rows_pnl(filtered, straddle_premium=prem, take_profit=tp)
summary = summarize_rows(rows, straddle_premium=prem, take_profit=tp)
if period == "custom" or str(period).startswith("custom:"):
period_label = period if str(period).startswith("custom:") else f"custom:{sample_days}"
else:
period_label = str(period)
miss = missing or []
return {
"ok": True,
"exchange": EXCHANGE,
"symbol": key,
"symbol_label": SYMBOLS[key]["label"],
"start_hour": sh,
"symbol": symbol_key,
"symbol_label": SYMBOLS[symbol_key]["label"],
"start_hour": start_hour,
"end_hour": END_HOUR,
"period": period_label,
"sample_days_requested": sample_days,
"straddle_premium": prem,
"take_profit": tp,
"weekend_filter": we_mode,
"timeframe": TIMEFRAME,
"price_source": price_source,
"inst_id": inst_id,
"timezone": "Asia/Shanghai",
"rows_all": rows_all,
"rows": rows,
"summary": summary,
"missing_days": missing[:30],
"missing_count": len(missing),
"missing_days": miss[:30],
"missing_count": len(miss),
}
def reframe_amp_stats(
*,
rows_all: list[dict[str, Any]],
symbol: str = "eth",
start_hour: int = 16,
period: str = "2m",
sample_days: int = 60,
straddle_premium: Any = None,
take_profit: Any = None,
weekend_filter: Any = "all",
price_source: str = "",
inst_id: str = "",
missing: Optional[list[str]] = None,
) -> dict[str, Any]:
"""已有日表上改周末/权利金/止盈,不拉 K 线."""
key = normalize_symbol(symbol)
return build_amp_result(
rows_all=list(rows_all or []),
symbol_key=key,
start_hour=int(start_hour),
period=period,
sample_days=int(sample_days or 60),
straddle_premium=straddle_premium,
take_profit=take_profit,
weekend_filter=weekend_filter,
price_source=price_source,
inst_id=inst_id,
missing=missing,
)
def rows_page(rows: list[dict[str, Any]], *, page: int = 1, page_size: int = 20) -> dict[str, Any]:
page = max(1, int(page or 1))
page_size = max(5, min(100, int(page_size or 20)))
@@ -470,6 +656,7 @@ def build_export_csv(payload: dict[str, Any]) -> str:
w.writerow(["起点整点", f"{payload.get('start_hour')}:00"])
w.writerow(["终点", f"{payload.get('end_hour')}:00"])
w.writerow(["周期", payload.get("period")])
w.writerow(["周末筛选", payload.get("weekend_filter")])
w.writerow(["样本数", s.get("sample_count")])
w.writerow(["最大振幅", s.get("max_amplitude"), "日期", s.get("max_amplitude_day")])
w.writerow(["振幅均值", s.get("avg_amplitude"), "中位数", s.get("median_amplitude")])
@@ -479,10 +666,12 @@ def build_export_csv(payload: dict[str, Any]) -> str:
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("premium"), "止盈点", st.get("take_profit")])
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")])
if st.get("take_profit") is not None:
w.writerow(["触达止盈天数", st.get("tp_hit_days"), "占比", st.get("tp_hit_ratio")])
w.writerow(
[
"买跨点数盈亏合计",
@@ -501,6 +690,8 @@ def build_export_csv(payload: dict[str, Any]) -> str:
w.writerow(
[
"结算日",
"星期",
"周末",
"窗起点",
"窗终点",
"开盘",
@@ -511,12 +702,17 @@ def build_export_csv(payload: dict[str, Any]) -> str:
"开→低",
"振幅",
"涨跌值",
"有效波动",
"触达止盈",
"收益",
]
)
for r in payload.get("rows") or []:
w.writerow(
[
r.get("settlement_day"),
r.get("weekday_label") or "",
"是" if r.get("is_weekend") else "否",
r.get("window_start"),
r.get("window_end"),
r.get("open"),
@@ -527,6 +723,9 @@ def build_export_csv(payload: dict[str, Any]) -> str:
r.get("down_points"),
r.get("amplitude"),
r.get("change"),
r.get("effective_move"),
"是" if r.get("take_profit_hit") else "否",
r.get("profit"),
]
)
return buf.getvalue()