Replace amp-stats straddle/perp overlays with move-points amplitude ratio.
Input points now drives amplitude hit share; table keeps both-side moves and amp达标. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
+24
-125
@@ -89,73 +89,22 @@ class AmpStatsLibTests(unittest.TestCase):
|
||||
self.assertEqual(s["max_amplitude_day"], "2026-07-02")
|
||||
self.assertEqual(s["max_up_points"], 500)
|
||||
self.assertEqual(s["max_down_points"], 200)
|
||||
self.assertIsNone(s["straddle"])
|
||||
self.assertIsNone(s["perp_hedge"])
|
||||
self.assertIsNone(s["move_points_stats"])
|
||||
|
||||
def test_perp_hedge_hit_and_pnl(self):
|
||||
from lib.hub.amp_stats_lib import perp_hedge_day_pnl, perp_hedge_day_pnl_eod
|
||||
|
||||
# 开盘=1800 optLev=100 → prem=36; A≈52.83; B=51
|
||||
# 触达目标出场 → 日盈亏=15, 不再按收盘涨跌算满仓
|
||||
def test_move_points_amp_ratio(self):
|
||||
rows = [
|
||||
{"open": 1800, "close": 1860, "change": 60, "up_points": 60, "down_points": 0, "amplitude": 60, "settlement_day": "2026-07-01"},
|
||||
{"open": 1800, "close": 1740, "change": -60, "up_points": 0, "down_points": 60, "amplitude": 60, "settlement_day": "2026-07-02"},
|
||||
{"open": 1800, "close": 1820, "change": 20, "up_points": 20, "down_points": 0, "amplitude": 20, "settlement_day": "2026-07-03"},
|
||||
{"amplitude": 100, "up_points": 40, "down_points": 60, "change": 10, "settlement_day": "2026-07-01"},
|
||||
{"amplitude": 40, "up_points": 10, "down_points": 30, "change": -5, "settlement_day": "2026-07-02"},
|
||||
{"amplitude": 50, "up_points": 50, "down_points": 0, "change": 20, "settlement_day": "2026-07-03"},
|
||||
]
|
||||
hedge = {
|
||||
"target_profit_u": 15,
|
||||
"perp_leverage": 10,
|
||||
"option_leverage": 100,
|
||||
"ratio_perp": 1,
|
||||
"ratio_opt": 2,
|
||||
"ct_mult": 0.01,
|
||||
}
|
||||
s = summarize_rows(rows, perp_hedge=hedge)
|
||||
ph = s["perp_hedge"]
|
||||
self.assertIsNotNone(ph)
|
||||
self.assertEqual(ph["entry"], "open")
|
||||
self.assertEqual(ph["exit"], "target_or_eod")
|
||||
self.assertEqual(ph["spot"], 1800.0)
|
||||
self.assertEqual(ph["opt_coins"], 2.0)
|
||||
self.assertEqual(ph["premium_total"], 36.0)
|
||||
self.assertAlmostEqual(ph["move_b"], 51.0, places=4)
|
||||
self.assertEqual(ph["hit_a_days"], 1) # up 60 >= A
|
||||
self.assertEqual(ph["hit_b_days"], 1) # down 60 >= B
|
||||
self.assertEqual(ph["target_exit_days"], 2)
|
||||
self.assertEqual(ph["eod_days"], 1)
|
||||
# 触达目标 → 15U
|
||||
self.assertAlmostEqual(ph["pnl_max"], 15.0, places=4)
|
||||
up_pnl = perp_hedge_day_pnl(
|
||||
change=60,
|
||||
open_px=1800,
|
||||
close_px=1860,
|
||||
option_leverage=100,
|
||||
opt_coins=2,
|
||||
up_points=60,
|
||||
down_points=0,
|
||||
target_profit_u=15,
|
||||
)
|
||||
down_pnl = perp_hedge_day_pnl(
|
||||
change=-60,
|
||||
open_px=1800,
|
||||
close_px=1740,
|
||||
option_leverage=100,
|
||||
opt_coins=2,
|
||||
up_points=0,
|
||||
down_points=60,
|
||||
target_profit_u=15,
|
||||
)
|
||||
self.assertAlmostEqual(up_pnl, 15.0, places=4)
|
||||
self.assertAlmostEqual(down_pnl, 15.0, places=4)
|
||||
self.assertAlmostEqual(ph["down_pnl_total"], 15.0, places=4)
|
||||
# 未触达:收盘结算
|
||||
eod = perp_hedge_day_pnl_eod(
|
||||
change=20, open_px=1800, close_px=1820, option_leverage=100, opt_coins=2
|
||||
)
|
||||
self.assertLess(eod, 0)
|
||||
self.assertEqual(ph["up_days"], 2)
|
||||
self.assertEqual(ph["down_days"], 1)
|
||||
|
||||
s = summarize_rows(rows, move_points=50)
|
||||
ms = s["move_points_stats"]
|
||||
self.assertIsNotNone(ms)
|
||||
self.assertEqual(ms["move_points"], 50)
|
||||
self.assertEqual(ms["amp_hit_days"], 2) # 100, 50
|
||||
self.assertEqual(ms["amp_hit_ratio"], round(2 / 3, 4))
|
||||
self.assertEqual(ms["up_hit_days"], 1) # 50
|
||||
self.assertEqual(ms["down_hit_days"], 1) # 60
|
||||
csv_text = build_export_csv(
|
||||
{
|
||||
"exchange": "okx",
|
||||
@@ -166,43 +115,12 @@ class AmpStatsLibTests(unittest.TestCase):
|
||||
"end_hour": 16,
|
||||
}
|
||||
)
|
||||
self.assertIn("永期对冲对照", csv_text)
|
||||
self.assertIn("永期盈亏", csv_text)
|
||||
self.assertIn("按日开盘", csv_text)
|
||||
self.assertIn("振幅占比", csv_text)
|
||||
self.assertIn("振幅达标", csv_text)
|
||||
|
||||
def test_long_straddle_stats(self):
|
||||
rows = [
|
||||
# |chg|=40>30 win+10; up=40>30; down=10
|
||||
{"up_points": 40, "down_points": 10, "change": 40, "amplitude": 50, "settlement_day": "2026-07-01"},
|
||||
# |chg|=10 lose-20; up=5; down=35>30
|
||||
{"up_points": 5, "down_points": 35, "change": -10, "amplitude": 40, "settlement_day": "2026-07-02"},
|
||||
# |chg|=30 not >30 lose-30; boundary
|
||||
{"up_points": 30, "down_points": 30, "change": 30, "amplitude": 60, "settlement_day": "2026-07-03"},
|
||||
]
|
||||
s = summarize_rows(rows, straddle_premium=30)
|
||||
st = s["straddle"]
|
||||
self.assertEqual(st["side"], "long_straddle")
|
||||
self.assertEqual(st["premium"], 30)
|
||||
self.assertEqual(st["up_exceed_days"], 1) # only 40
|
||||
self.assertEqual(st["down_exceed_days"], 1) # only 35
|
||||
self.assertEqual(st["abs_change_exceed_days"], 1) # only 40
|
||||
self.assertAlmostEqual(st["pnl_total"], 40 - 30 + 10 - 30 + 30 - 30)
|
||||
self.assertEqual(st["win_days"], 1)
|
||||
self.assertEqual(st["win_ratio"], round(1 / 3, 4))
|
||||
csv_text = build_export_csv(
|
||||
{"exchange": "okx", "symbol_label": "ETH", "summary": s, "rows": rows, "start_hour": 22, "end_hour": 16}
|
||||
)
|
||||
self.assertIn("买跨对照", csv_text)
|
||||
self.assertIn("买跨点数盈亏合计", csv_text)
|
||||
def test_weekend_and_reframe_move_points(self):
|
||||
from lib.hub.amp_stats_lib import enrich_rows, filter_weekend_rows, reframe_amp_stats
|
||||
|
||||
def test_take_profit_and_weekend(self):
|
||||
from lib.hub.amp_stats_lib import (
|
||||
enrich_rows_pnl,
|
||||
filter_weekend_rows,
|
||||
reframe_amp_stats,
|
||||
)
|
||||
|
||||
# Sat 2026-07-18, Sun 2026-07-19, Mon 2026-07-20
|
||||
rows = [
|
||||
{
|
||||
"settlement_day": "2026-07-18",
|
||||
@@ -238,39 +156,20 @@ class AmpStatsLibTests(unittest.TestCase):
|
||||
only = filter_weekend_rows(rows, "only")
|
||||
self.assertEqual(len(only), 2)
|
||||
|
||||
# TP=80: day1 hit → move 80; day2 no → |12|; day3 no → 8
|
||||
enriched = enrich_rows_pnl(rows, straddle_premium=10, take_profit=80)
|
||||
self.assertTrue(enriched[0]["take_profit_hit"])
|
||||
self.assertEqual(enriched[0]["effective_move"], 80)
|
||||
self.assertEqual(enriched[0]["profit"], 70)
|
||||
self.assertFalse(enriched[1]["take_profit_hit"])
|
||||
self.assertEqual(enriched[1]["effective_move"], 12)
|
||||
self.assertEqual(enriched[1]["profit"], 2)
|
||||
# TP empty → use |change|
|
||||
no_tp = enrich_rows_pnl(rows[:1], straddle_premium=10, take_profit=None)
|
||||
self.assertEqual(no_tp[0]["effective_move"], 5)
|
||||
self.assertEqual(no_tp[0]["profit"], -5)
|
||||
|
||||
# TP boundary >= : up=80 counts as hit
|
||||
edge = enrich_rows_pnl(
|
||||
[{"up_points": 80, "down_points": 1, "change": 2, "settlement_day": "2026-07-20", "is_weekend": False}],
|
||||
straddle_premium=10,
|
||||
take_profit=80,
|
||||
)
|
||||
self.assertTrue(edge[0]["take_profit_hit"])
|
||||
self.assertEqual(edge[0]["profit"], 70)
|
||||
enriched = enrich_rows(rows, move_points=80)
|
||||
self.assertTrue(enriched[0]["amp_hit"])
|
||||
self.assertFalse(enriched[1]["amp_hit"])
|
||||
self.assertTrue(enriched[2]["amp_hit"])
|
||||
|
||||
reframed = reframe_amp_stats(
|
||||
rows_all=rows,
|
||||
symbol="eth",
|
||||
weekend_filter="exclude",
|
||||
straddle_premium=10,
|
||||
take_profit=80,
|
||||
move_points=80,
|
||||
)
|
||||
self.assertEqual(reframed["summary"]["sample_count"], 1)
|
||||
# Mon: 未触达止盈 → |8|-10
|
||||
self.assertEqual(reframed["rows"][0]["profit"], -2)
|
||||
self.assertIn("收益", build_export_csv(reframed))
|
||||
self.assertTrue(reframed["rows"][0]["amp_hit"])
|
||||
self.assertIn("振幅占比", build_export_csv(reframed))
|
||||
|
||||
def test_fetch_switches_to_history_endpoint(self):
|
||||
"""近期接口到头后应切 history 续拉."""
|
||||
|
||||
Reference in New Issue
Block a user