Update app.py

app.py

@@ -17,11 +17,11 @@ except ImportError:
 
 from flask import Flask, request, jsonify
 
-# db_signals helpers you already have; we add read_user_json_file below.
+# db_signals helpers you already have
 from db_signals import (
     fetch_authenticity_token_and_commit_oid,
     update_user_json_file,
-    # If you already have a reader, import it instead of defining here.
+    get_user_json_file,  # we will call this directly
 )
 
 # ================ Configuration ================
@@ -73,17 +73,23 @@ ACTIVE_MONITOR = {
     "entry": None,
     "tp": None,
     "sl": None,
-    "timestamps": None  # keep activation timestamps to preserve on /ts_points updates
+    "timestamps": None  # preserved for /ts_points
 }
 ACTIVE_LOCK = threading.Lock()
 
+# New: single activator governance
+ACTIVE_ACTIVATOR = {
+    "thread": None,
+    "cancel_token": None,
+    "id": None
+}
+ACTIVATOR_LOCK = threading.Lock()
+
+# Latch to prevent duplicate activations
+ACTIVATION_LATCH = threading.Event()
+
 # ================ db_signals reader helper ================
-# If you already have an official reader, replace this function with it.
 def read_user_json_file(authenticity_token: str, commit_oid: str):
-    """
-    Returns parsed JSON: list or dict, or [] if empty/unparseable.
-    """
-    from db_signals import get_user_json_file
     res = get_user_json_file(authenticity_token, commit_oid)
     if not res.get("success"):
         raise RuntimeError(f"Failed to read user json file: {res}")
@@ -176,7 +182,7 @@ def send_message_to_users(message: str, max_retries=5, retry_delay=10):
             response = requests.post(MESSAGE_API_URL, headers=headers, data=json.dumps(payload), timeout=8)
             if response.status_code == 200:
                 logger.info(f"Message sent to users successfully on attempt {attempt}")
-                return {"success": True, "response": response.json()}
+                return {"success": True, "response": response.json() if response.content else {}}
             else:
                 logger.warning(f"Attempt {attempt}: Users API status {response.status_code}, body: {response.text[:200]}")
         except requests.exceptions.RequestException as e:
@@ -252,10 +258,10 @@ def call_analysis_now(text: str):
     except Exception as e:
         logger.warning(f"[ANALYSIS] analysis_now failed: {e}")
 
-# ================ GitHub signal writers/readers ================
+# ================ GitHub writers =================
 def write_active_scenario_to_github(pair: str, side: str, entry: str, sl: str, tp: str, dt_utc: datetime.datetime, preserve_timestamps: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
     """
-    Always writes signals.json as a single-element array [ { active_trade } ].
+    Writes signals.json as a single-element array [ { active_trade } ].
     If preserve_timestamps provided, keep it; else compute fresh timestamps.
     """
     timestamps = preserve_timestamps if preserve_timestamps else format_zone_times(dt_utc)
@@ -289,7 +295,7 @@ def write_active_scenario_to_github(pair: str, side: str, entry: str, sl: str, t
 
 def clear_github_signals_to_empty_list() -> Dict[str, Any]:
     """
-    Clears the db_signals file to [] after SL/TP hit as requested.
+    Clears the db_signals file to [] after SL/TP hit.
     """
     try:
         authenticity_token, commit_oid = fetch_authenticity_token_and_commit_oid()
@@ -304,19 +310,14 @@ def clear_github_signals_to_empty_list() -> Dict[str, Any]:
         logger.error(f"[GITHUB] clear exception: {e}")
         return {"success": False, "message": str(e)}
 
-def read_current_scenario_from_github() -> Optional[Dict[str, Any]]:
+# ================ Scenario fetcher ================
+def fetch_current_scenario_struct() -> Optional[Dict[str, Any]]:
     """
-    Reads the pre-activation scenario JSON when signals.json is an array, e.g.:
-    [
-      {
-        "scenario": {
-          "Buy":  {"at": "...", "SL": "...", "TP": "..."},
-          "Sell": {"at": "...", "SL": "...", "TP": "..."},
-          "timestamps": {...}
-        }
-      }
-    ]
-    Returns {"scenario": {...}} or None if not found.
+    Reads the pre-activation scenario JSON when signals.json is an array:
+    [ { "scenario": { "Buy": {"at": "...", "SL": "...", "TP": "..."},
+                      "Sell": {"at": "...", "SL": "...", "TP": "..."},
+                      "timestamps": {...} } } ]
+    Returns {"scenario": {...}} or None.
     """
     try:
         authenticity_token, commit_oid = fetch_authenticity_token_and_commit_oid()
@@ -344,7 +345,6 @@ def read_current_scenario_from_github() -> Optional[Dict[str, Any]]:
     except Exception as e:
         logger.warning(f"[SCENARIO] fetch failed: {e}")
         return None
-        
 
 # ================ Activation helpers ================
 class CancelToken:
@@ -499,6 +499,7 @@ def start_sl_tp_monitor(side: str, entry: float, sl: float, tp: float, symbol: s
 
     t = threading.Thread(target=_runner, daemon=True)
     with ACTIVE_LOCK:
+        # Cancel any running monitor before starting a new one
         if ACTIVE_MONITOR["cancel_token"]:
             ACTIVE_MONITOR["cancel_token"].cancel()
         ACTIVE_MONITOR["thread"] = t
@@ -509,25 +510,9 @@ def start_sl_tp_monitor(side: str, entry: float, sl: float, tp: float, symbol: s
         ACTIVE_MONITOR["entry"] = entry
         ACTIVE_MONITOR["tp"] = tp
         ACTIVE_MONITOR["sl"] = sl
-        # timestamps preserved in ACTIVE_MONITOR["timestamps"] by activator
+        # timestamps remains as is (set by activator)
     t.start()
 
-# ================ Scenario fetcher ================
-def fetch_current_scenario_struct() -> Optional[Dict[str, Any]]:
-    """
-    Reads the pre-activation scenario (Buy/Sell at/SL/TP) from db_signals storage.
-    Returns:
-      {
-        "scenario": {
-          "Buy": {"at": "...", "SL": "...", "TP": "..."},
-          "Sell": {"at": "...", "SL": "...", "TP": "..."},
-          "timestamps": {...}
-        }
-      }
-      or None if not available.
-    """
-    return read_current_scenario_from_github()
-
 # ================ API Endpoints ================
 @app.route("/track", methods=["POST"])
 def track_signal():
@@ -536,11 +521,11 @@ def track_signal():
     Behavior:
       - Stop analysis and clear other alert (immediately).
       - Cancel any existing activation/monitor (newest wins).
-      - Start watcher for activation of Buy/Sell (gap-aware). On first hit:
+      - Start watcher for activation of Buy/Sell. On first hit (latched):
           * Play sound
-          * Read SL/TP from scenario in db_signals
+          * Use cached SL/TP from scenario read at start
           * Write active array [ {...} ] with timestamps for the start moment
-          * Send user "BUY NOW/SELL NOW..." message
+          * Send user "BUY NOW/SELL NOW..." message (once)
           * Start SL/TP monitor
           * Notify analysis via analysis_now
       - Respond 200 immediately after accepting.
@@ -571,11 +556,10 @@ def track_signal():
     # 1) Stop external analysis and other alert now
     call_stop_analysis_and_clear_others()
 
-    # 2) Cancel any current operation
+    # 2) Cancel any current SL/TP monitor operation and reset active monitor state
     with ACTIVE_LOCK:
         if ACTIVE_MONITOR["cancel_token"]:
             ACTIVE_MONITOR["cancel_token"].cancel()
-        # Reset active monitor state; activation not yet started
         ACTIVE_MONITOR["thread"] = None
         ACTIVE_MONITOR["cancel_token"] = None
         ACTIVE_MONITOR["id"] = None
@@ -586,75 +570,118 @@ def track_signal():
         ACTIVE_MONITOR["sl"] = None
         ACTIVE_MONITOR["timestamps"] = None
 
-    activator_cancel = CancelToken()
-
-    def _activator():
-        try:
-            act = wait_until_activation_dual(
-                buy_entry=buy_at,
-                sell_entry=sell_at,
-                symbol=SYMBOL_DEFAULT,
-                cancel_token=activator_cancel,
-                max_minutes=240
-            )
-            if act is None:
-                logger.info("[ACTIVATOR] Cancelled or timed out.")
-                return
-
-            chosen_side = act["side"]
-            entry_used = act["price"]
-
-            play_alert_sound()
+    # 3) Cancel any existing activator and set up new one
+    with ACTIVATOR_LOCK:
+        if ACTIVE_ACTIVATOR["cancel_token"]:
+            ACTIVE_ACTIVATOR["cancel_token"].cancel()
+        ACTIVATION_LATCH.clear()  # new cycle, no activation yet
 
-            # Read scenario SL/TP from db_signals
-            scenario_blob = fetch_current_scenario_struct()
-            if not scenario_blob or "scenario" not in scenario_blob:
-                logger.error("[ACTIVATOR] Missing scenario data to read SL/TP.")
-                send_message_to_users(f"ERROR: Missing scenario SL/TP for {chosen_side} at {entry_used}")
-                return
+        activator_cancel = CancelToken()
+        activator_id = f"act-{int(time.time()*1000)}-{random.randint(1000,9999)}"
 
-            side_block = scenario_blob["scenario"].get(chosen_side, {})
+        def _activator():
             try:
-                sl_f = float(side_block.get("SL"))
-                tp_f = float(side_block.get("TP"))
-            except Exception:
-                logger.error("[ACTIVATOR] Invalid SL/TP from scenario data.")
-                send_message_to_users(f"ERROR: Invalid SL/TP for {chosen_side} at {entry_used}")
-                return
+                # Read scenario once and cache SL/TP locally to avoid race
+                scenario_blob = fetch_current_scenario_struct()
+                if not scenario_blob or "scenario" not in scenario_blob:
+                    logger.error("[ACTIVATOR] Missing scenario data before waiting.")
+                    send_message_to_users("ERROR: Missing scenario structure before activation. Please set scenario and retry.")
+                    return
 
-            # 3) Write active array with fresh timestamps for the start moment
-            dt_utc = now_utc()
-            ts_block = format_zone_times(dt_utc)
-            wr = write_active_scenario_to_github(
-                pair=SYMBOL_DEFAULT,
-                side=chosen_side,
-                entry=str(entry_used),
-                sl=str(sl_f),
-                tp=str(tp_f),
-                dt_utc=dt_utc
-            )
-            if not wr.get("success"):
-                logger.warning(f"[GITHUB] Write failed: {wr}")
-
-            # Save timestamps in memory to preserve on /ts_points updates
-            with ACTIVE_LOCK:
-                ACTIVE_MONITOR["timestamps"] = ts_block
+                scen = scenario_blob["scenario"]
+                # Extract SL/TP for both sides if present
+                buy_sl = buy_tp = sell_sl = sell_tp = None
+                if "Buy" in scen:
+                    try:
+                        buy_sl = float(scen["Buy"].get("SL")) if scen["Buy"].get("SL") is not None else None
+                        buy_tp = float(scen["Buy"].get("TP")) if scen["Buy"].get("TP") is not None else None
+                    except Exception:
+                        logger.error("[ACTIVATOR] Invalid Buy SL/TP in scenario.")
+                if "Sell" in scen:
+                    try:
+                        sell_sl = float(scen["Sell"].get("SL")) if scen["Sell"].get("SL") is not None else None
+                        sell_tp = float(scen["Sell"].get("TP")) if scen["Sell"].get("TP") is not None else None
+                    except Exception:
+                        logger.error("[ACTIVATOR] Invalid Sell SL/TP in scenario.")
+
+                # Optional timestamps from scenario pre-activation are not used here; we create fresh on activate
+
+                act = wait_until_activation_dual(
+                    buy_entry=buy_at,
+                    sell_entry=sell_at,
+                    symbol=SYMBOL_DEFAULT,
+                    cancel_token=activator_cancel,
+                    max_minutes=240
+                )
+                if act is None:
+                    logger.info("[ACTIVATOR] Cancelled or timed out.")
+                    return
 
-            # 4) Notify users
-            user_msg = f"{chosen_side.upper()} NOW at {entry_used}\nSL: {sl_f}\nTP: {tp_f}\nPair: {SYMBOL_DEFAULT}"
-            send_message_to_users(user_msg)
+                # Latch: only the first activator proceeds
+                if ACTIVATION_LATCH.is_set():
+                    logger.info("[ACTIVATOR] Activation already handled by another thread. Exiting.")
+                    return
+                ACTIVATION_LATCH.set()
 
-            # 5) Inform analysis
-            call_analysis_now(f"Scenario {chosen_side} activated for {SYMBOL_DEFAULT} at {entry_used}. SL {sl_f}, TP {tp_f}.")
+                chosen_side = act["side"]
+                entry_used = act["price"]
 
-            # 6) Start SL/TP monitor
-            start_sl_tp_monitor(side=chosen_side, entry=entry_used, sl=sl_f, tp=tp_f, symbol=SYMBOL_DEFAULT)
+                # Choose SL/TP from cached scenario
+                if chosen_side == "Buy":
+                    sl_f, tp_f = buy_sl, buy_tp
+                else:
+                    sl_f, tp_f = sell_sl, sell_tp
 
-        except Exception as e:
-            logger.error(f"[ACTIVATOR] Error: {e}")
+                if sl_f is None or tp_f is None:
+                    # Cannot proceed without points
+                    logger.error(f"[ACTIVATOR] Missing SL/TP for {chosen_side}.")
+                    send_message_to_users(f"ERROR: Missing scenario SL/TP for {chosen_side} at {entry_used}")
+                    return
 
-    t = threading.Thread(target=_activator, daemon=True)
-    t.start()
+                play_alert_sound()
+
+                # 3) Write active array with fresh timestamps
+                dt_utc = now_utc()
+                ts_block = format_zone_times(dt_utc)
+                wr = write_active_scenario_to_github(
+                    pair=SYMBOL_DEFAULT,
+                    side=chosen_side,
+                    entry=str(entry_used),
+                    sl=str(sl_f),
+                    tp=str(tp_f),
+                    dt_utc=dt_utc
+                )
+                if not wr.get("success"):
+                    logger.warning(f"[GITHUB] Write failed: {wr}")
+
+                # Save timestamps in memory to preserve on /ts_points updates
+                with ACTIVE_LOCK:
+                    ACTIVE_MONITOR["timestamps"] = ts_block
+
+                # 4) Notify users (once)
+                user_msg = f"{chosen_side.upper()} NOW at {entry_used}\nSL: {sl_f}\nTP: {tp_f}\nPair: {SYMBOL_DEFAULT}"
+                send_message_to_users(user_msg)
+
+                # 5) Inform analysis
+                call_analysis_now(f"Scenario {chosen_side} activated for {SYMBOL_DEFAULT} at {entry_used}. SL {sl_f}, TP {tp_f}.")
+
+                # 6) Start SL/TP monitor
+                start_sl_tp_monitor(side=chosen_side, entry=entry_used, sl=sl_f, tp=tp_f, symbol=SYMBOL_DEFAULT)
+
+            except Exception as e:
+                logger.error(f"[ACTIVATOR] Error: {e}")
+            finally:
+                with ACTIVATOR_LOCK:
+                    if ACTIVE_ACTIVATOR["id"] == activator_id:
+                        ACTIVE_ACTIVATOR["thread"] = None
+                        ACTIVE_ACTIVATOR["cancel_token"] = None
+                        ACTIVE_ACTIVATOR["id"] = None
+
+        t = threading.Thread(target=_activator, daemon=True)
+        ACTIVE_ACTIVATOR["thread"] = t
+        ACTIVE_ACTIVATOR["cancel_token"] = activator_cancel
+        ACTIVE_ACTIVATOR["id"] = activator_id
+        t.start()
 
     return jsonify({"status": "accepted", "message": f"Tracking activation for Buy at {buy_at} and/or Sell at {sell_at} started."}), 200
 
@@ -708,7 +735,7 @@ def ts_points():
     # Restart monitor with new TP/SL
     start_sl_tp_monitor(side=side, entry=entry, sl=sl, tp=tp, symbol=symbol)
 
-    # Update GitHub record preserving timestamps (do NOT change time)
+    # Update GitHub record preserving timestamps
     wr = write_active_scenario_to_github(
         pair=symbol,
         side=side,
@@ -721,6 +748,13 @@ def ts_points():
     if not wr.get("success"):
         logger.warning(f"[GITHUB] Update after TS change failed: {wr}")
 
+    # Notify users and analysis about the update (optional; you can enable if needed)
+    try:
+        send_message_to_users(f"{side.upper()} TP/SL updated\nEntry: {entry}\nTP: {tp}\nSL: {sl}\nPair: {symbol}")
+        call_analysis_now(f"{symbol} {side} TP/SL updated. Entry {entry}, TP {tp}, SL {sl}.")
+    except Exception:
+        pass
+
     return jsonify({"status": "ok", "message": "TP/SL updated and tracking restarted"}), 200
 
 @app.route("/", methods=["GET"])
@@ -733,10 +767,11 @@ def root():
             "POST /ts_points": "Payload: { 'TP': new_tp?, 'SL': new_sl? }. Updates active trade without changing start time."
         },
         "notes": [
-            "Only one operation active at a time; newest request cancels previous (across users).",
-            "On activation: stop analysis & clear other alert, write single-scenario array with fresh timestamps, notify user, start SL/TP monitor, notify analysis.",
-            "On SL/TP hit: notify user first, then clear db_signals to [], stop analysis & clear other alert, and inform analysis.",
-            "On /ts_points: do NOT change timestamps in GitHub; preserve trade start time; restart monitor with new TP/SL."
+            "Only one activator and one monitor run at a time; newest /track cancels previous.",
+            "Activation uses a latch so only one activation proceeds even under concurrency.",
+            "Scenario SL/TP are cached at activator start; no GitHub reads at activation moment.",
+            "On SL/TP hit: notify user, clear db_signals to [], stop analysis & clear other alert, and inform analysis.",
+            "On /ts_points: timestamps in GitHub are preserved (trade start time)."
         ]
     }), 200