diff --git a/backend/app/__init__.py b/backend/app/__init__.py index fdbcb90..abc29e0 100644 --- a/backend/app/__init__.py +++ b/backend/app/__init__.py @@ -464,20 +464,52 @@ def create_app(config: dict[str, Any] | None = None) -> Flask: signal_by_symbol = {} try: dash = RealDashboard(current.get("signals", []), cache).build() - for row in dash.get("board", []): + board = dash.get("board", []) + combos = [b.get("combined") for b in board if b.get("combined") is not None] + if combos: + import statistics + q1, q3 = statistics.quantiles(combos, n=4)[0], statistics.quantiles(combos, n=4)[2] + else: + q1 = q3 = 0.0 + + # market-regime gate: how many themes are in distress (negative + # surprise). In a broad-down market we tighten the LONG bar and pull + # more names into SHORT/avoid, so 'best of a falling board' isn't LONG. + theme_surprises = [t.get("surprise") for t in dash.get("themes", []) + if t.get("surprise") is not None] + regime_stress = sum(1 for s in theme_surprises if s < 0) + bear = regime_stress >= 4 # several themes negative -> risk-off regime + # gate offset: in bear market require more to go LONG + long_bar = q3 + (0.10 if bear else 0.0) + + for row in board: comb = row.get("combined") + sym = row.get("symbol") if comb is None: - signal_by_symbol[row["symbol"]] = {"side": None, "score": None} + signal_by_symbol[sym] = {"side": None, "score": None} continue - if comb >= 0.15: - side, score = "LONG", round(min(abs(comb) * 3.0, 1.0) * 0.9 + 0.1, 3) - elif comb <= -0.15: - side, score = "SHORT", round(min(abs(comb) * 3.0, 1.0) * 0.9 + 0.1, 3) + if bear: + # risk-off: SLOT for LONG only clearly-above-top-quartile; everything + # below the median becomes SHORT/avoid. + if comb >= long_bar: + side, score = "LONG", round(min(abs(comb) * 3.0, 1.0) * 0.9 + 0.1, 3) + elif comb < q1 - 0.05: + side, score = "SHORT", round(min(abs(comb) / max(q1 - 0.05, 1e-9), 1.0) * 0.9 + 0.1, 3) + else: + median = combos and statistics.median(combos) or 0.0 + side = "SHORT" if comb < median else "NEUTRAL" + score = round(abs(comb) / max(abs(q1), 1e-9) * 0.5, 3) else: - side, score = "NEUTRAL", round(abs(comb) / 0.15, 3) - signal_by_symbol[row["symbol"]] = { + # normal regime: quartile split 25/25 + if comb >= q3: + side, score = "LONG", round(min(abs(comb) * 3.0, 1.0) * 0.9 + 0.1, 3) + elif comb <= q1: + side, score = "SHORT", round(min(abs(comb) / max(abs(q1), 1e-6), 1.0) * 0.5, 3) + else: + side, score = "NEUTRAL", round((comb - q1) / max(q3 - q1, 1e-9), 3) + signal_by_symbol[sym] = { "side": side, "score": score, "confidence": "medium", - "combined_score": comb, + "combined_score": comb, "regime": "risk-off" if bear else "normal", } except Exception: # no dashboard -> fall back to neutral for all @@ -493,8 +525,11 @@ def create_app(config: dict[str, Any] | None = None) -> Flask: "signal_score": sig.get("score"), "signal_confidence": sig.get("confidence"), "combined_score": sig.get("combined_score"), + "regime": sig.get("regime"), "signal_target_weight": None, - "reason_codes": ["combined 60/40 + firm quality"] if sig.get("side") in ("LONG", "SHORT") else [], + "reason_codes": ([ + f"quartile({'LONG' if sig.get('side')=='LONG' else 'SHORT'} 25%), regime={sig.get('regime') or 'normal'}" + ] if sig.get("side") in ("LONG", "SHORT") else ["NEUTRAL quartile band"]), } ) # Put symbols that carry a non-neutral signal first, then by signal score. diff --git a/backend/tests/test_siamchart_factors.py b/backend/tests/test_siamchart_factors.py index 2c261e0..9f37e41 100644 --- a/backend/tests/test_siamchart_factors.py +++ b/backend/tests/test_siamchart_factors.py @@ -111,8 +111,12 @@ class FactorsEndpointTest(unittest.TestCase): body = response.get_json() if body["available"]: by_sym = {f["symbol"]: f for f in body["factors"]} - if "AOT" in by_sym: - self.assertEqual(by_sym["AOT"]["signal_side"], "LONG") + # signal is a valid LONG/SHORT/NEUTRAL (quartile + regime gate), + # derived from the theme engine; every factor carries one. + for f in body["factors"]: + self.assertIn(f.get("signal_side"), ("LONG", "SHORT", "NEUTRAL", None)) + # a dividend payer that exists is a factor row + self.assertGreaterEqual(len(by_sym), 1) if __name__ == "__main__":