import json import os import tempfile import unittest from unittest import mock import server class TrendBreakReadyDashboardTests(unittest.TestCase): def _write_results(self, tmpdir, signals): with open(os.path.join(tmpdir, "trend_break_results.json"), "w", encoding="utf-8") as handle: json.dump({ "timestamp_iso": "2026-08-21T09:00:00+07:00", "source": "test", "snapshot_phase": "LIVE", "market_open": True, "universe_count": 100, "candidate_count": len(signals), "signals": signals, }, handle) def _signal(self, status="CONFIRM"): return { "ticker": "TEST", "status": status, "screening_status": "KANDIDAT FULL PA", "score": 82.5, "price": 110, "detected_price": 108, "swing_high": 109, "break_margin_pct": 0.9, "vol_ratio": 2.1, "detected_at": "2026-08-21T09:00:00+07:00", "detected_date": "2026-08-21", "detected_time": "09:00:00", "screening_reasons": [], } def test_confirm_is_not_ready_without_pa_confirmation(self): with tempfile.TemporaryDirectory() as tmpdir: self._write_results(tmpdir, [self._signal()]) with ( mock.patch.object(server.os, "getcwd", return_value=tmpdir), mock.patch.object(server, "load_vcp_pa_status_payload", return_value={"statuses": {}}) as pa_status, mock.patch.object(server, "_load_trend_break_support_snapshot", return_value={"status": "ok", "rows": []}), mock.patch.object(server, "trend_break_live_payload", return_value={"data_age_sec": 10, "interval_sec": 60}), ): payload = server.load_trend_break_ready_snapshot() self.assertEqual(payload["counts"]["confirmed"], 1) self.assertEqual(payload["counts"]["ready"], 0) self.assertFalse(payload["rows"][0]["entry_ready"]) pa_status.assert_called_once_with(["TEST"]) def test_only_closed_full_pa_plan_becomes_ready(self): with tempfile.TemporaryDirectory() as tmpdir: self._write_results(tmpdir, [self._signal()]) pa_payload = { "statuses": { "TEST": { "status": "SIAP ENTRY", "reason": "closed 5m + volume + RR valid", "planning_only": False, "entry": 110, "sl": 100, "target": 130, "rr": 2.0, }, }, "as_of": "2026-08-21T09:01:00+07:00", "market_open": True, } with ( mock.patch.object(server.os, "getcwd", return_value=tmpdir), mock.patch.object(server, "load_vcp_pa_status_payload", return_value=pa_payload), mock.patch.object(server, "_load_trend_break_support_snapshot", return_value={"status": "ok", "rows": []}), mock.patch.object(server, "trend_break_live_payload", return_value={"data_age_sec": 10, "interval_sec": 60}), ): payload = server.load_trend_break_ready_snapshot() row = payload["rows"][0] self.assertEqual(payload["counts"]["ready"], 1) self.assertTrue(row["entry_ready"]) self.assertEqual(row["entry_status"], "SIAP ENTRY") self.assertEqual(row["plan"]["risk_pct"], 9.090909090909092) self.assertEqual(row["plan"]["reward_pct"], 18.181818181818183) def test_near_rows_are_visible_but_never_promoted(self): with tempfile.TemporaryDirectory() as tmpdir: self._write_results(tmpdir, [self._signal("NEAR")]) with ( mock.patch.object(server.os, "getcwd", return_value=tmpdir), mock.patch.object(server, "load_vcp_pa_status_payload") as pa_status, mock.patch.object(server, "_load_trend_break_support_snapshot", return_value={"status": "ok", "rows": []}), mock.patch.object(server, "trend_break_live_payload", return_value={"data_age_sec": 10, "interval_sec": 60}), ): payload = server.load_trend_break_ready_snapshot() self.assertEqual(payload["counts"]["near"], 1) self.assertEqual(payload["counts"]["confirmed"], 0) self.assertEqual(payload["rows"][0]["entry_status"], "BELUM CONFIRM") pa_status.assert_not_called() def test_support_ready_and_waiting_rows_join_without_promoting_unready_stock(self): with tempfile.TemporaryDirectory() as tmpdir: self._write_results(tmpdir, [self._signal("NEAR")]) support_rows = [ { "ticker": "AKRA", "phase": "SUPPORT KUAT", "screening_status": "KANDIDAT FULL PA", "swing_action_status": "MENUNGGU KONFIRMASI", "swing_action_reason": "Menunggu bias 1D dan konfirmasi 15m.", "close": 1395, "support_level": 1375, "support_zone_low": 1370, "support_zone_high": 1385, "touches_support": 3, "rejection_count": 2, "quality_score": 82, "volume_ratio_20d": 1.4, "retrieved_at": "2026-08-21T07:51:32+07:00", "data_through": "2026-08-20", "swing_plan": { "buy_price": 1395, "stop_loss": 1375, "take_profit": 1520, "risk_reward": 5.57, }, "strategy_entry": { "ready": False, "entry_status": "", "planning_only": True, "reason": "Menunggu bias 1D + setup 1H + konfirmasi 15m selaras.", }, }, { "ticker": "BBNI", "phase": "SUPPORT KUAT", "screening_status": "KANDIDAT FULL PA", "swing_action_status": "SETUP SWING TERKONFIRMASI", "close": 5000, "support_level": 4900, "quality_score": 88, "volume_ratio_20d": 1.8, "retrieved_at": "2026-08-21T07:51:32+07:00", "strategy_entry": { "ready": True, "method": "SUPPORT_REJECTION", "entry_status": "SIAP ENTRY SESI BERIKUTNYA", "planning_only": True, "reason": "Bullish rejection closed 5m dengan volume mendukung.", "plan": { "entry": 5000, "sl": 4900, "target": 5300, "rr": 3.0, "risk_pct": 2.0, "reward_pct": 6.0, }, }, }, { "ticker": "UNRELATED", "phase": "SUPPORT KUAT", "screening_status": "TIDAK LAYAK DITUNGGU", "swing_action_status": "SIAP ENTRY", "strategy_entry": {"ready": True, "entry_status": "SIAP ENTRY"}, }, ] with ( mock.patch.object(server.os, "getcwd", return_value=tmpdir), mock.patch.object(server, "load_vcp_pa_status_payload", return_value={"statuses": {}}), mock.patch.object( server, "_load_trend_break_support_snapshot", return_value={ "status": "ok", "rows": support_rows, "as_of": "2026-08-21T07:51:32+07:00", "market_open": False, "watchlist": "ALL_WATCHLISTS", "data_source": "test support", }, ), mock.patch.object(server, "trend_break_live_payload", return_value={"data_age_sec": 10, "interval_sec": 60}), ): payload = server.load_trend_break_ready_snapshot() self.assertEqual(payload["counts"]["support_rows"], 2) self.assertEqual(payload["counts"]["support_candidates"], 2) self.assertEqual(payload["counts"]["waiting_confirmation"], 1) self.assertEqual(payload["counts"]["ready"], 1) ready = payload["ready"][0] self.assertEqual(ready["source"], "SUPPORT REACTION") self.assertEqual(ready["entry_status"], "SIAP ENTRY SESI BERIKUTNYA") self.assertEqual(ready["entry_method"], "SUPPORT_REJECTION") self.assertEqual(payload["counts"]["queue_rows"], 3) if __name__ == "__main__": unittest.main()