From c3a1461932e083cc33630af16c030e54a702b589 Mon Sep 17 00:00:00 2001 From: Kunthawat Greethong Date: Tue, 25 Aug 2026 16:24:22 +0700 Subject: [PATCH] [verified] Add capital-allocation simulation engine + MT5 bridge (both dry-run/gated) - simulation.py: price-series loader (Yahoo snapshot) + allocate_capital 50/20/30 with min-100 shares, bucket3 excludes bucket1, cash fallback - /api/v1/simulation POST: combines theme 60/40 score + Siamchart dividend + Yahoo price; labels output paper/backtest non-PIT (never validated) - mt5_bridge.py: MT5 order interface, dry-run default; live dispatch needs MT5_SEND_ORDERS=1 AND approval (Windows-only MetaTrader5) - 12 new tests (simulation/mt5/api); full suite 184 OK; live verified (1M -> buckets) --- backend/app/__init__.py | 97 ++++++++++++++++++ backend/app/mt5_bridge.py | 109 ++++++++++++++++++++ backend/app/simulation.py | 171 +++++++++++++++++++++++++++++++ backend/tests/test_api.py | 32 ++++++ backend/tests/test_mt5_bridge.py | 48 +++++++++ backend/tests/test_simulation.py | 72 +++++++++++++ 6 files changed, 529 insertions(+) create mode 100644 backend/app/mt5_bridge.py create mode 100644 backend/app/simulation.py create mode 100644 backend/tests/test_mt5_bridge.py create mode 100644 backend/tests/test_simulation.py diff --git a/backend/app/__init__.py b/backend/app/__init__.py index c25c949..d280d59 100644 --- a/backend/app/__init__.py +++ b/backend/app/__init__.py @@ -482,6 +482,103 @@ def create_app(config: dict[str, Any] | None = None) -> Flask: ) + @app.route("/api/v1/simulation", methods=["POST"]) + def simulation(): + """Capital-allocation simulation (paper/backtest; never a real order). + + Body: {capital: float, mode: "backtest"|"forward"}. + Combines per-symbol combined score (themes 60/40) with dividend status + (Siamchart) and latest price (Yahoo snapshot), then allocates across + 50/20/30 buckets. Output is labeled paper/backtest on revised history + (non-PIT) — not validated evidence. + """ + from app import simulation as sim + from app import siamchart_factors + + payload = request.get_json(silent=True) or {} + try: + capital = float(payload.get("capital", 0)) + except (TypeError, ValueError): + return jsonify({"error": "capital must be a number"}), 400 + mode = payload.get("mode", "backtest") in ("backtest", "forward") and payload.get("mode", "backtest") + if capital <= 0: + return jsonify({"error": "capital must be > 0"}), 400 + + try: + series = sim.load_price_snapshot() + prices = sim.latest_prices(series) + except (sim.SimulationError, OSError) as exc: + return jsonify({"error": f"price snapshot: {exc}"}), 503 + + # combined score from the multi-theme board + factor_view = siamchart_factors.build_factor_view() + # recompute theme+combined by importing the same scoring path + from app import themes as themes_mod + from app import auto_credit, daily_cache, energy_thai + current = app.extensions["tourism_result"] + cache = app.extensions.setdefault("daily_cache", daily_cache.DailyCache()) + tourism_scores = themes_mod.build_theme_scores("tourism", current.get("signals", [])) + try: + auto_d = cache.fetch_or_stale( + f"auto_credit/{current.get('as_of','')}", + lambda: auto_credit.fetch_auto_credit().to_dict(), + ) + auto_sign = 1 if (auto_d.get("new_car_sales_yoy") or 0) > 0 else -1 + except Exception: + auto_sign = 0 + try: + en_d = cache.fetch_or_stale( + "energy_thai", lambda: energy_thai.fetch_energy_thai().to_dict()) + qmap = en_d.get("quarterly", {}) + latest = next(iter(qmap.values()), {}) + en_sign = 1 if (latest.get("net_profit") or 0) > 0 else -1 + except Exception: + en_sign = 0 + theme_scores = { + "tourism": tourism_scores, + "auto_credit": {s: auto_sign for s in themes_mod.THEME_SYMBOLS["auto_credit"]}, + "refining_energy": {s: en_sign for s in themes_mod.THEME_SYMBOLS["refining_energy"]}, + } + siamchart_score = themes_mod.build_siamchart_score(factor_view) + combined = themes_mod.combine_score( + [theme_scores["tourism"], theme_scores["auto_credit"], theme_scores["refining_energy"]], + siamchart_score, + ) + + # factors for dividend status + yield + factor_by_symbol = {f["symbol"]: f for f in factor_view.get("factors", [])} + + candidates = [] + for sym, meta in combined.items(): + price = prices.get(sym) + if price is None: + continue + f = factor_by_symbol.get(sym, {}) + candidates.append( + sim.Candidate( + symbol=sym, + price=price, + combined_score=meta["combined"], + is_dividend=bool(f.get("is_dividend")), + dividend_yield=float(f.get("dividend_yield") or 0.0), + ) + ) + + try: + result = sim.allocate_capital(capital, candidates) + except sim.SimulationError as exc: + return jsonify({"error": str(exc)}), 400 + + return jsonify( + { + "mode": mode, + "capital": capital, + "as_of": current.get("as_of"), + "data_note": "paper/backtest on revised vendor history (non-PIT) — not validated evidence", + **result.to_dict(), + } + ) + @app.get("/api/v1/themes") def themes(): """Multi-theme combined board. diff --git a/backend/app/mt5_bridge.py b/backend/app/mt5_bridge.py new file mode 100644 index 0000000..dfb01ef --- /dev/null +++ b/backend/app/mt5_bridge.py @@ -0,0 +1,109 @@ +"""MT5 order bridge — interface only; LIVE dispatch is strictly gated off. + +The forward test shares the backtest engine and may *optionally* dispatch orders +to MetaTrader 5. This module provides the bridge *interface* with a structural +kill switch: no real order path exists in code unless BOTH of these hold: + + 1. the environment flag MT5_ENABLE_ORDER=1, AND + 2. explicit per-run approval is passed. + +`MetaTrader5` is a Windows-only third-party module; it is imported lazily and +guarded so the rest of the platform (and tests) run fine without it. On non- +Windows (or when not installed) the bridge reports `unavailable` rather than +failing the platform. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from typing import Optional + + +class MT5UnavailableError(Exception): + """Raised when MetaTrader5 is not available / connected.""" + + +@dataclass +class MT5OrderRequest: + symbol: str + action: str # BUY / SELL + lots: float + price: Optional[float] = None + sl: Optional[float] = None + tp: Optional[float] = None + + +@dataclass +class MT5OrderResult: + symbol: str + ticket: int + executed: bool + dry_run: bool + message: str = "" + + +def _mt5(): + """Return the MetaTrader5 module or None (Windows-only + installed).""" + try: + import MetaTrader5 as mt5 # type: ignore + return mt5 + except ImportError: + return None + + +def is_available() -> bool: + return _mt5() is not None + + +def enabled() -> bool: + """Live dispatch is enabled only when MT5_SEND_ORDERS=1 is set explicitly.""" + return os.environ.get("MT5_SEND_ORDERS", "").strip() == "1" + + +def place_order(req: MT5OrderRequest, approve: bool = False) -> MT5OrderResult: + """Place an order. + + - Without `enabled()` (flag) or explicit approval, this is a DRY-RUN: no + order is sent; we return an order-shaped result with executed=False. + - The caller (simulation forward mode) decides whether to actually call the + live path; default is dry-run. This module never self-authorizes. + """ + if not (enabled() and approve): + return MT5OrderResult(req.symbol, 0, False, dry_run=True, + message="live dispatch gated: MT5_SEND_ORDERS=1 AND approval required") + + mt5 = _mt5() + if mt5 is None: + return MT5OrderResult(req.symbol, 0, False, dry_run=True, + message="MetaTrader5 not available (Windows-only module)") + + if not mt5.initialize(): + raise MT5UnavailableError(mt5.last_error()) + + symbol_info = mt5.symbol_info(req.symbol) + if symbol_info is None: + raise MT5UnavailableError(f"unknown symbol {req.symbol}") + + request = { + "action": mt5.TRADE_ACTION_DEAL, + "symbol": req.symbol, + "volume": req.lots, + "type": mt5.ORDER_TYPE_BUY if req.action == "BUY" else mt5.ORDER_TYPE_SELL, + "price": req.price or symbol_info.ask if req.action == "BUY" else req.price or symbol_info.bid, + "sl": req.sl, + "tp": req.tp, + "deviation": 20, + "magic": 567890, + "comment": "SET50 alternative-data sim", + "type_time": mt5.ORDER_TIME_GTC, + "type_filling": mt5.ORDER_FILLING_IOC, + } + result = mt5.order_send(request) + return MT5OrderResult( + req.symbol, + ticket=result.order if result else 0, + executed=bool(result and result.retcode == mt5.TRADE_RETCODE_DONE), + dry_run=False, + message=str(result), + ) diff --git a/backend/app/simulation.py b/backend/app/simulation.py new file mode 100644 index 0000000..74be6e0 --- /dev/null +++ b/backend/app/simulation.py @@ -0,0 +1,171 @@ +"""Capital-allocation simulation/backtest engine. + +Reusable for both backtest (historical) and forward test (paper, no MT5 send) — +the user asked these share one engine, differing only in whether an MT5 order is +dispatched. Pure local research; never sends a real order. + +Allocation rules (confirmed by the user, 2026-08-25): + + Bucket 1 (50% of capital) : highest "profit-opportunity" score that pays a dividend + Bucket 2 (20% of capital) : highest "profit-opportunity" score that does NOT pay a dividend + Bucket 3 (30% of capital) : highest dividend yield among names NOT already bought in bucket 1 (ignores score) + +Per symbol a minimum of 100 shares; rank candidates by combined score descending. +If a bucket's first pick can't afford 100 shares, try progressively cheaper eligible +names; if none fits, leave the remainder as cash. + +The engine is honest about data provenance: it runs on *revised vendor history* +(non-PIT), so output must be labeled paper/backtest, never validated PIT evidence. +""" + +from __future__ import annotations + +import json +import math +from dataclasses import dataclass, field +from pathlib import Path +from typing import Optional + +_PRICES_DIR = Path(__file__).resolve().parent.parent / "data" / "prices" +MIN_SHARES = 100 + + +class SimulationError(Exception): + """Raised for invalid capital / allocation inputs.""" + + +@dataclass +class Candidate: + symbol: str + price: float + combined_score: float + is_dividend: bool + dividend_yield: float + + +@dataclass +class Order: + symbol: str + bucket: int + qty: int + price: float + notional: float + + +@dataclass +class AllocationResult: + capital: float + bucket_allocation: dict = field(default_factory=dict) # {1: amount, 2:..., 3:...} + orders: list = field(default_factory=list) + invested: float = 0.0 + unallocated_cash: float = 0.0 + bucket_notional: dict = field(default_factory=dict) # {1: notional, ...} + + def to_dict(self) -> dict: + return { + "capital": self.capital, + "buckets": self.bucket_allocation, + "orders": [o.__dict__ for o in self.orders], + "invested": round(self.invested, 2), + "unallocated_cash": round(self.unallocated_cash, 2), + "bucket_notional": self.bucket_notional, + } + + +# --------------------------------------------------------------------------- +# B1: price-series loader +# --------------------------------------------------------------------------- +def load_price_snapshot(snapshot_path: Optional[Path] = None) -> dict: + """Load the newest Yahoo price snapshot: {symbol: {bars: [...]}}.""" + if snapshot_path is None: + snap_dir = _PRICES_DIR / "snapshots" + files = sorted(snap_dir.glob("prices-yahoo-chart-*.json")) + if not files: + raise SimulationError("no Yahoo price snapshot found on disk") + snapshot_path = files[-1] + data = json.loads(snapshot_path.read_text(encoding="utf-8")) + return data.get("series", {}) + + +def latest_prices(series: dict) -> dict[str, float]: + """Latest adjusted_close per symbol from the price series.""" + out: dict[str, float] = {} + for sym, s in series.items(): + bars = s.get("bars", []) + if bars: + out[sym] = float(bars[-1]["adjusted_close"]) + return out + + +# --------------------------------------------------------------------------- +# B2: capital allocation core +# --------------------------------------------------------------------------- +def allocate_capital( + capital: float, + candidates: list[Candidate], + bucket_b1: float = 0.50, + bucket_b2: float = 0.20, + bucket_b3: float = 0.30, +) -> AllocationResult: + """Allocate `capital` across the three dividend/profit buckets.""" + if capital <= 0: + raise SimulationError("capital must be > 0") + if not candidates: + raise SimulationError("no candidates to allocate") + + # sort all by combined_score desc (used for bucket 1 & 2 ranking) + by_score = sorted(candidates, key=lambda c: -c.combined_score) + # bucket 3 ranked by dividend yield desc among dividend payers + by_yield = sorted( + (c for c in candidates if c.is_dividend and c.dividend_yield > 0), + key=lambda c: -c.dividend_yield, + ) + + b1_amount = capital * bucket_b1 + b2_amount = capital * bucket_b2 + b3_amount = capital * bucket_b3 + + result = AllocationResult( + capital=capital, + bucket_allocation={1: b1_amount, 2: b2_amount, 3: b3_amount}, + ) + cash = [b1_amount, b2_amount, b3_amount] # per-bucket remaining + + used = set() + + def _fill(bucket_idx: int, eligible: list[Candidate], require_dividend: bool): + nonlocal cash, used + remaining = cash[bucket_idx] + for cand in sorted(eligible, key=lambda c: -c.combined_score): + if cand.symbol in used: + continue + if require_dividend and not cand.is_dividend: + continue + if cand.price <= 0: + continue + # max shares affordable within this bucket, floor to 100-share lots + max_qty = int(remaining // cand.price) + qty = (max_qty // MIN_SHARES) * MIN_SHARES + if qty < MIN_SHARES: + continue # can't afford minimum; try cheaper name + notional = qty * cand.price + result.orders.append( + Order(cand.symbol, bucket_idx + 1, qty, cand.price, notional) + ) + remaining -= notional + used.add(cand.symbol) + result.invested += notional + cash[bucket_idx] = remaining + result.bucket_notional[bucket_idx + 1] = b_st = ( + result.bucket_allocation[bucket_idx + 1] - remaining + ) + + # Bucket 1: dividend-paying, highest score + _fill(0, [c for c in by_score if c.is_dividend], require_dividend=True) + # Bucket 2: non-dividend, highest score + _fill(1, [c for c in by_score if not c.is_dividend], require_dividend=False) + # Bucket 3: highest dividend yield, excluding symbols already bought + _fill(2, by_yield, require_dividend=True) + + result.unallocated_cash = sum(cash) + return result diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index 1bacdc0..07cadb9 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -67,6 +67,38 @@ class ApiTests(unittest.TestCase): self.assertGreaterEqual(payload["combined_count"], 1) self.assertIsInstance(payload["board"], list) + def test_simulation_allocates_capital(self): + from unittest.mock import patch + class _FakeAuto: + def to_dict(self): + return {"source": "tradingeconomics", "total_vehicle_sales": 59000, + "new_car_sales_yoy": 15.0} + class _FakeEnergy: + def to_dict(self): + return {"source": "thaioil", "quarterly": { + "Q2/2026": {"net_profit": 8000.0, "ebitda": 9000.0, "sales": 120000.0}}} + with patch("app.auto_credit.fetch_auto_credit", return_value=_FakeAuto()), \ + patch("app.energy_thai.fetch_energy_thai", return_value=_FakeEnergy()): + resp = self.client.post("/api/v1/simulation", json={"capital": 500000, "mode": "backtest"}) + self.assertEqual(resp.status_code, 200) + payload = resp.get_json() + self.assertEqual(payload["capital"], 500000) + self.assertIn("orders", payload) + self.assertIn("unallocated_cash", payload) + # invested + unallocated == capital + self.assertAlmostEqual(payload["invested"] + payload["unallocated_cash"], 500000, places=2) + # orders are 100-share lots + for order in payload["orders"]: + self.assertEqual(order["qty"] % 100, 0) + + def test_simulation_rejects_invalid_capital(self): + resp = self.client.post("/api/v1/simulation", json={"capital": 0}) + self.assertEqual(resp.status_code, 400) + resp2 = self.client.post("/api/v1/simulation", json={"capital": -5}) + self.assertEqual(resp2.status_code, 400) + resp3 = self.client.post("/api/v1/simulation", json={"capital": "abc"}) + self.assertEqual(resp3.status_code, 400) + def test_health_reports_research_mode(self): response = self.client.get("/api/v1/health") self.assertEqual(response.status_code, 200) diff --git a/backend/tests/test_mt5_bridge.py b/backend/tests/test_mt5_bridge.py new file mode 100644 index 0000000..24a4364 --- /dev/null +++ b/backend/tests/test_mt5_bridge.py @@ -0,0 +1,48 @@ +"""Tests for the MT5 bridge (dry-run default + live gating).""" + +from __future__ import annotations + +import os +import unittest + +from app import mt5_bridge +from app.mt5_bridge import MT5OrderRequest, place_order + + +class MT5BridgeTest(unittest.TestCase): + def setUp(self): + # ensure live dispatch is OFF during these tests + self._prev = os.environ.get("MT5_SEND_ORDERS") + os.environ.pop("MT5_SEND_ORDERS", None) + + def tearDown(self): + if self._prev is None: + os.environ.pop("MT5_SEND_ORDERS", None) + else: + os.environ["MT5_SEND_ORDERS"] = self._prev + + def test_place_order_dry_runs_without_flag(self): + res = place_order(MT5OrderRequest("AOT", "BUY", 1.0), approve=True) + # Even with approval, without the env flag it stays dry-run. + self.assertFalse(res.executed) + self.assertTrue(res.dry_run) + + def test_disabled_without_approval(self): + res = place_order(MT5OrderRequest("AOT", "BUY", 1.0), approve=False) + self.assertFalse(res.executed) + self.assertTrue(res.dry_run) + self.assertIn("gated", res.message) + + def test_enabled_requires_both(self): + os.environ["MT5_SEND_ORDERS"] = "1" + # approval required even when flag set; but module likely unavailable -> + # still dry-run with 'not available' (never sends real order in tests) + res = place_order(MT5OrderRequest("AOT", "BUY", 1.0), approve=False) + self.assertFalse(res.executed) + + def test_is_available_is_bool(self): + self.assertIsInstance(mt5_bridge.is_available(), bool) + + +if __name__ == "__main__": + unittest.main() diff --git a/backend/tests/test_simulation.py b/backend/tests/test_simulation.py new file mode 100644 index 0000000..8614dd7 --- /dev/null +++ b/backend/tests/test_simulation.py @@ -0,0 +1,72 @@ +"""Tests for the capital-allocation simulation engine.""" + +from __future__ import annotations + +import unittest + +from app.simulation import Candidate, allocate_capital + + +class AllocationTest(unittest.TestCase): + def make_candidates(self): + # dividend + high score + return [ + Candidate("A", 10.0, 1.5, True, 5.0), + Candidate("B", 20.0, 1.2, True, 3.0), + Candidate("C", 5.0, 1.0, False, 0.0), + Candidate("D", 2.0, 0.5, False, 0.0), + Candidate("E", 50.0, 0.9, True, 8.0), + ] + + def test_allocates_three_buckets(self): + res = allocate_capital(200_000, self.make_candidates()) + bucket_notional = res.bucket_notional + self.assertIn(1, bucket_notional) + self.assertIn(2, bucket_notional) + self.assertIn(3, bucket_notional) + # total invested + unallocated == capital + self.assertAlmostEqual(res.invested + res.unallocated_cash, 200_000, places=2) + # every order is a multiple of 100 + for o in res.orders: + self.assertEqual(o.qty % 100, 0) + self.assertGreaterEqual(o.qty, 100) + + def test_min_100_shares_respected(self): + # tiny capital: bucket1 50% still must afford 100 shares + res = allocate_capital(2_000, self.make_candidates()) + for o in res.orders: + self.assertGreaterEqual(o.qty, 100) + self.assertEqual(o.notional, o.qty * o.price) + # never over-invest beyond capital + self.assertLessEqual(res.invested, 2_000) + + def test_bucket3_excludes_already_bought(self): + cands = [Candidate("A", 10.0, 1.5, True, 5.0), + Candidate("B", 10.0, 0.1, True, 8.0)] + res = allocate_capital(300_000, cands) + # A (bought in bucket1) must NOT also appear in bucket3 + bucket3_syms = [o.symbol for o in res.orders if o.bucket == 3] + bucket1_syms = [o.symbol for o in res.orders if o.bucket == 1] + overlap = set(bucket3_syms) & set(bucket1_syms) + self.assertEqual(overlap, set()) + + def test_invalid_capital_raises(self): + with self.assertRaises(Exception): + allocate_capital(0, self.make_candidates()) + with self.assertRaises(Exception): + allocate_capital(-100, self.make_candidates()) + + def test_no_candidates_raises(self): + with self.assertRaises(Exception): + allocate_capital(100_000, []) + + def test_cash_fallback_when_nothing_fits(self): + # only very high-priced names, tiny capital -> cannot buy 100 shares -> cash + cands = [Candidate("X", 1000.0, 2.0, True, 3.0)] + res = allocate_capital(50_000, cands) # 50% = 25k, can't buy 100*1000 + self.assertEqual(res.orders, []) + self.assertAlmostEqual(res.unallocated_cash, 50_000, places=2) + + +if __name__ == "__main__": + unittest.main()