🚀 Auto-deploy: Add background automated scheduler, Kelly allocation, multi-portfolio simulations, and UI improvements

app.py

@@ -20,6 +20,27 @@ from agents.papert import PapertAgent
 from agents.audit import AuditAgent
 
 app = Flask(__name__)
+
+# ── SQLite helpers with WAL mode and error resilience ───────────────────────
+def _db_conn(db_path):
+    """Get a WAL-mode connection with busy timeout."""
+    try:
+        conn = sqlite3.connect(db_path, timeout=30)
+        conn.execute("PRAGMA journal_mode=DELETE")
+        conn.execute("PRAGMA busy_timeout=30000")
+        conn.execute("PRAGMA synchronous=NORMAL")
+        return conn
+    except Exception as e:
+        app.logger.error(f"DB connect error ({db_path}): {e}")
+        raise
+
+def _safe_commit(conn, label=""):
+    """Commit with error handling."""
+    try:
+        conn.commit()
+    except Exception as e:
+        app.logger.error(f"DB commit error {label}: {e}")
+
 app.secret_key = os.getenv("SECRET_KEY", "brainsteel-fin-secret-2025")
 
 # Paths
@@ -34,7 +55,7 @@ AGENTS_DB = os.path.join(DATA_DIR, "agents.db")
 
 # ── Databases ─────────────────────────────────────────────────────────────────
 def init_db():
-    conn = sqlite3.connect(DB_PATH)
+    conn = _db_conn(DB_PATH)
     c = conn.cursor()
     c.execute("""
         CREATE TABLE IF NOT EXISTS executions (
@@ -70,12 +91,12 @@ def init_db():
             cost_usd REAL
         )
     """)
-    conn.commit()
+    _safe_commit(conn)
     conn.close()
 
 def init_agents_db():
     """Agent registry — onboarding + state."""
-    conn = sqlite3.connect(AGENTS_DB)
+    conn = _db_conn(AGENTS_DB)
     c = conn.cursor()
     c.execute("""
         CREATE TABLE IF NOT EXISTS agents (
@@ -117,12 +138,12 @@ def init_agents_db():
             metadata TEXT
         )
     """)
-    conn.commit()
+    _safe_commit(conn)
     conn.close()
 
 def seed_agents():
     """Seed default agents on first run."""
-    conn = sqlite3.connect(AGENTS_DB)
+    conn = _db_conn(AGENTS_DB)
     c = conn.cursor()
     # Check if already seeded
     row = c.execute("SELECT COUNT(*) FROM agents").fetchone()[0]
@@ -145,7 +166,7 @@ def seed_agents():
         for a in defaults:
             c.execute("INSERT INTO agents (id,name,role,intent,model,provider,status,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?)",
                       (*a, now, now))
-        conn.commit()
+        _safe_commit(conn)
     conn.close()
 
 init_db()
@@ -197,21 +218,21 @@ state = AgentState()
 
 # ── Logging ───────────────────────────────────────────────────────────────────
 def log_execution(agent, action, result, status="success"):
-    conn = sqlite3.connect(DB_PATH)
+    conn = _db_conn(DB_PATH)
     c = conn.cursor()
     c.execute("INSERT INTO executions (timestamp, agent, action, result, status) VALUES (?, ?, ?, ?, ?)",
               (datetime.now().isoformat(), agent, action, result, status))
-    conn.commit()
+    _safe_commit(conn)
     conn.close()
     # Also log to agent_logs
     _log_agent_event(agent, "execution", result)
 
 def _log_agent_event(agent_id, event, result, duration_ms=0):
-    conn = sqlite3.connect(AGENTS_DB)
+    conn = _db_conn(AGENTS_DB)
     c = conn.cursor()
     c.execute("INSERT INTO agent_logs (agent_id, timestamp, event, duration_ms, result) VALUES (?, ?, ?, ?, ?)",
               (agent_id, datetime.now().isoformat(), event, duration_ms, result))
-    conn.commit()
+    _safe_commit(conn)
     conn.close()
 
 # ── Pipeline Execution ────────────────────────────────────────────────────────
@@ -233,7 +254,6 @@ def run_pipeline():
         brief_data = brief.run()
         t1 = datetime.now()
         state.update("brief", "done", brief_data.get("summary", "")[:80])
-        # Also persist brief key metrics to state for /api/state
         brief_summary = brief_data.get("summary", "")
         state.brief.update({
             "price": brief_data.get("price"),
@@ -255,8 +275,12 @@ def run_pipeline():
             "macd_histogram": brief_data.get("macd_histogram"),
             "btc_dominance": brief_data.get("btc_dominance"),
         })
-        log_execution("brief", "market_analysis", brief_summary, "success")
-        _update_agent_stats("brief", t1 - t0, True)
+        # RESILIENT: Skip DB write if fails
+        try:
+            log_execution("brief", "market_analysis", brief_summary, "success")
+            _update_agent_stats("brief", t1 - t0, True)
+        except Exception as dbg:
+            app.logger.warning(f"DB write skipped (non-fatal): {dbg}")
 
         # ── Step 2: Decio ──
         t0 = datetime.now()
@@ -272,8 +296,11 @@ def run_pipeline():
             "take_profit": decio_data.get("take_profit"),
             "justification": decio_data.get("justification", "")[:100],
         })
-        log_execution("decio", "strategy_decision", decio_data.get("decision", "HOLD"), "success")
-        _update_agent_stats("decio", t1 - t0, True)
+        try:
+            log_execution("decio", "strategy_decision", decio_data.get("decision", "HOLD"), "success")
+            _update_agent_stats("decio", t1 - t0, True)
+        except Exception as dbg:
+            app.logger.warning(f"DB write skipped (non-fatal): {dbg}")
 
         # ── Step 3: PaperT ──
         t0 = datetime.now()
@@ -282,8 +309,11 @@ def run_pipeline():
         papert_data = papert.run()
         t1 = datetime.now()
         state.update("papert", "done", papert_data.get("result", "")[:80])
-        log_execution("papert", "order_execution", papert_data.get("result", ""), "success")
-        _update_agent_stats("papert", t1 - t0, True)
+        try:
+            log_execution("papert", "order_execution", papert_data.get("result", ""), "success")
+            _update_agent_stats("papert", t1 - t0, True)
+        except Exception as dbg:
+            app.logger.warning(f"DB write skipped (non-fatal): {dbg}")
 
         # ── Step 4: Audit ──
         t0 = datetime.now()
@@ -292,31 +322,40 @@ def run_pipeline():
         audit_data = audit.audit_pipeline(brief_data, decio_data, papert_data)
         t1 = datetime.now()
         state.update("audit", "done", audit_data.get("summary", "")[:80])
-        log_execution("audit", "compliance_check", audit_data.get("summary", ""), "success")
-        _update_agent_stats("audit", t1 - t0, True)
+        try:
+            log_execution("audit", "compliance_check", audit_data.get("summary", ""), "success")
+            _update_agent_stats("audit", t1 - t0, True)
+        except Exception as dbg:
+            app.logger.warning(f"DB write skipped (non-fatal): {dbg}")
 
         # ── Save pipeline run ──
-        conn = sqlite3.connect(DB_PATH)
-        c = conn.cursor()
-        c.execute("""
-            INSERT INTO pipeline_runs (started_at, finished_at, brief_result, decio_decision, papert_result, audit_report, status)
-            VALUES (?, ?, ?, ?, ?, ?, ?)
-        """, (
-            start.isoformat(),
-            datetime.now().isoformat(),
-            brief_data.get("summary", ""),
-            decio_data.get("decision", "HOLD"),
-            papert_data.get("result", ""),
-            json.dumps(audit_data),
-            "success"
-        ))
-        conn.commit()
-        conn.close()
+        try:
+            conn = _db_conn(DB_PATH)
+            c = conn.cursor()
+            c.execute("""
+                INSERT INTO pipeline_runs (started_at, finished_at, brief_result, decio_decision, papert_result, audit_report, status)
+                VALUES (?, ?, ?, ?, ?, ?, ?)
+            """, (
+                start.isoformat(),
+                datetime.now().isoformat(),
+                brief_data.get("summary", ""),
+                decio_data.get("decision", "HOLD"),
+                papert_data.get("result", ""),
+                json.dumps(audit_data),
+                "success"
+            ))
+            _safe_commit(conn)
+            conn.close()
+        except Exception as dbg:
+            app.logger.warning(f"Pipeline run log skipped (non-fatal): {dbg}")
 
     except Exception as e:
-        log_execution("pipeline", "error", str(e), "error")
+        app.logger.error(f"Pipeline error: {e}")
+        try:
+            log_execution("pipeline", "error", str(e), "error")
+        except:
+            pass
         state.update("brief", "error", f"Erro: {str(e)[:80]}")
-        _update_agent_stats("brief", datetime.now() - start, False)
 
     finally:
         state.set_pipeline_running(False)
@@ -324,8 +363,10 @@ def run_pipeline():
         state.update("papert", "idle", "Aguardando...")
         state.update("audit", "idle", "Aguardando...")
 
+    return brief_data, decio_data, papert_data, audit_data
+
 def _update_agent_stats(agent_id, duration, success):
-    conn = sqlite3.connect(AGENTS_DB)
+    conn = _db_conn(AGENTS_DB)
     c = conn.cursor()
     now = datetime.now().isoformat()
     dur = int(duration.total_seconds() * 1000)
@@ -342,7 +383,7 @@ def _update_agent_stats(agent_id, duration, success):
         count = row[1]
         new_avg = int((old_avg * (count - 1) + dur) / count)
         c.execute("UPDATE agents SET avg_duration_ms=? WHERE id=?", (new_avg, agent_id))
-    conn.commit()
+    _safe_commit(conn)
     conn.close()
 
 # ── Routes ────────────────────────────────────────────────────────────────────
@@ -364,7 +405,7 @@ def api_run():
 
 @app.route("/api/history")
 def api_history():
-    conn = sqlite3.connect(DB_PATH)
+    conn = _db_conn(DB_PATH)
     conn.row_factory = sqlite3.Row
     c = conn.cursor()
     rows = c.execute("SELECT * FROM executions ORDER BY timestamp DESC LIMIT 50").fetchall()
@@ -373,7 +414,7 @@ def api_history():
 
 @app.route("/api/pipeline_history")
 def api_pipeline_history():
-    conn = sqlite3.connect(DB_PATH)
+    conn = _db_conn(DB_PATH)
     conn.row_factory = sqlite3.Row
     c = conn.cursor()
     rows = c.execute("SELECT * FROM pipeline_runs ORDER BY started_at DESC LIMIT 50").fetchall()
@@ -391,7 +432,7 @@ def api_token_stats():
 @app.route("/api/token_totals", methods=["GET"])
 def api_token_totals():
     """All-time token totals per agent."""
-    conn = sqlite3.connect(DB_PATH)
+    conn = _db_conn(DB_PATH)
     c = conn.cursor()
     c.execute("""
         SELECT agent, SUM(prompt_tokens) as p, SUM(completion_tokens) as c,
@@ -407,7 +448,7 @@ def api_token_totals():
 @app.route("/api/agents", methods=["GET"])
 def get_agents():
     """List all registered agents with full status."""
-    conn = sqlite3.connect(AGENTS_DB)
+    conn = _db_conn(AGENTS_DB)
     conn.row_factory = sqlite3.Row
     c = conn.cursor()
     rows = c.execute("SELECT * FROM agents ORDER BY id").fetchall()
@@ -416,7 +457,7 @@ def get_agents():
 
 @app.route("/api/agents/<agent_id>", methods=["GET"])
 def get_agent(agent_id):
-    conn = sqlite3.connect(AGENTS_DB)
+    conn = _db_conn(AGENTS_DB)
     conn.row_factory = sqlite3.Row
     c = conn.cursor()
     row = c.execute("SELECT * FROM agents WHERE id=?", (agent_id,)).fetchone()
@@ -429,7 +470,7 @@ def get_agent(agent_id):
 def update_agent(agent_id):
     """Update agent config (model, provider, status, intent)."""
     data = request.json
-    conn = sqlite3.connect(AGENTS_DB)
+    conn = _db_conn(AGENTS_DB)
     c = conn.cursor()
     fields = []
     values = []
@@ -442,18 +483,18 @@ def update_agent(agent_id):
         values.append(datetime.now().isoformat())
         values.append(agent_id)
         c.execute(f"UPDATE agents SET {','.join(fields)} WHERE id=?", values)
-        conn.commit()
+        _safe_commit(conn)
     conn.close()
     return jsonify({"status": "updated", "id": agent_id})
 
 @app.route("/api/agents/<agent_id>", methods=["DELETE"])
 def delete_agent(agent_id):
     """Remove agent from registry."""
-    conn = sqlite3.connect(AGENTS_DB)
+    conn = _db_conn(AGENTS_DB)
     c = conn.cursor()
     c.execute("DELETE FROM agent_logs WHERE agent_id=?", (agent_id,))
     c.execute("DELETE FROM agents WHERE id=?", (agent_id,))
-    conn.commit()
+    _safe_commit(conn)
     conn.close()
     return jsonify({"status": "deleted", "id": agent_id})
 
@@ -463,7 +504,7 @@ def register_agent():
     data = request.json
     if not data.get("id") or not data.get("name"):
         return jsonify({"error": "id and name required"}), 400
-    conn = sqlite3.connect(AGENTS_DB)
+    conn = _db_conn(AGENTS_DB)
     c = conn.cursor()
     now = datetime.now().isoformat()
     c.execute("""
@@ -475,7 +516,7 @@ def register_agent():
         data.get("provider", "unknown"), data.get("status"),
         now, now
     ))
-    conn.commit()
+    _safe_commit(conn)
     conn.close()
     return jsonify({"status": "registered", "id": data["id"]}), 201
 
@@ -483,7 +524,7 @@ def register_agent():
 @app.route("/api/agent_metrics", methods=["GET"])
 def get_agent_metrics():
     """Dashboard metrics for all agents."""
-    conn = sqlite3.connect(AGENTS_DB)
+    conn = _db_conn(AGENTS_DB)
     conn.row_factory = sqlite3.Row
     c = conn.cursor()
     rows = c.execute("SELECT * FROM agents ORDER BY id").fetchall()
@@ -516,7 +557,7 @@ def get_agent_metrics():
 @app.route("/api/agent_logs/<agent_id>", methods=["GET"])
 def get_agent_logs(agent_id):
     """Event log for specific agent."""
-    conn = sqlite3.connect(AGENTS_DB)
+    conn = _db_conn(AGENTS_DB)
     conn.row_factory = sqlite3.Row
     c = conn.cursor()
     limit = request.args.get("limit", 50, type=int)
@@ -531,7 +572,7 @@ def get_agent_logs(agent_id):
 @app.route("/api/services", methods=["GET"])
 def get_services():
     """List all registered services."""
-    conn = sqlite3.connect(AGENTS_DB)
+    conn = _db_conn(AGENTS_DB)
     conn.row_factory = sqlite3.Row
     c = conn.cursor()
     rows = c.execute("SELECT * FROM services ORDER BY name").fetchall()
@@ -546,7 +587,7 @@ def register_service():
         return jsonify({"error": "name required"}), 400
     
     service_id = data.get("id") or hashlib.md5(data["name"].encode()).hexdigest()[:12]
-    conn = sqlite3.connect(AGENTS_DB)
+    conn = _db_conn(AGENTS_DB)
     c = conn.cursor()
     c.execute("""
         INSERT OR REPLACE INTO services (id, name, type, endpoint, status, last_check, metadata)
@@ -560,7 +601,7 @@ def register_service():
         datetime.now().isoformat(),
         json.dumps(data.get("metadata", {}))
     ))
-    conn.commit()
+    _safe_commit(conn)
     conn.close()
     return jsonify({"status": "registered", "id": service_id}), 201
 
@@ -568,7 +609,7 @@ def register_service():
 def ping_service(service_id):
     """Health check a service endpoint."""
     import requests as req
-    conn = sqlite3.connect(AGENTS_DB)
+    conn = _db_conn(AGENTS_DB)
     c = conn.cursor()
     row = c.execute("SELECT endpoint FROM services WHERE id=?", (service_id,)).fetchone()
     conn.close()
@@ -582,11 +623,11 @@ def ping_service(service_id):
     except Exception as e:
         status = "unreachable"
     
-    conn = sqlite3.connect(AGENTS_DB)
+    conn = _db_conn(AGENTS_DB)
     c = conn.cursor()
     c.execute("UPDATE services SET status=?, last_check=? WHERE id=?", 
                (status, datetime.now().isoformat(), service_id))
-    conn.commit()
+    _safe_commit(conn)
     conn.close()
     return jsonify({"id": service_id, "status": status, "endpoint": endpoint})
 
@@ -594,7 +635,7 @@ def ping_service(service_id):
 @app.route("/api/pipeline/visual")
 def pipeline_visual():
     """Real-time pipeline flow with metrics."""
-    conn = sqlite3.connect(DB_PATH)
+    conn = _db_conn(DB_PATH)
     conn.row_factory = sqlite3.Row
     c = conn.cursor()
     
@@ -698,5 +739,243 @@ def reset_portfolio():
     except Exception as e:
         return jsonify({"error": str(e)}), 500
 
+# ── Backtest API (Máquina do Tempo) ───────────────────────────────────────────
+@app.route("/api/backtest", methods=["POST"])
+def run_backtest():
+    """Motor de Backtesting Automatizado (Máquina do Tempo)."""
+    sys.path.insert(0, "/app/agents")
+    try:
+        import requests as req
+        from agents.decio import DecioAgent
+        from agents.papert import PapertAgent, Portfolio, PORTFOLIO_DB
+        from pathlib import Path
+
+        data = request.json or {}
+        days = data.get("days", 30)
+        initial_capital = data.get("initial_capital", 10000.0)
+
+        # Busca dados históricos reais da Binance (klines 1d)
+        r = req.get(f"https://api.binance.com/api/v3/klines?symbol=BTCUSDT&interval=1d&limit={days}", timeout=10)
+        if r.status_code != 200:
+            return jsonify({"error": "Falha ao buscar klines da Binance"}), 502
+
+        klines = r.json()
+
+        # Usa um banco de dados de backtest isolado
+        import sqlite3
+        backtest_db = "/app/data/backtest.db"
+        if os.path.exists(backtest_db):
+            os.remove(backtest_db)
+
+        # Sobrescreve PORTFOLIO_DB temporariamente para o backtest
+        import agents.papert as papert_module
+        old_db = papert_module.PORTFOLIO_DB
+        old_init = papert_module.INITIAL_BALANCE
+        papert_module.PORTFOLIO_DB = Path(backtest_db)
+        papert_module.INITIAL_BALANCE = initial_capital
+
+        pf_soberana = Portfolio(strategy="soberana")
+        pf_agressiva = Portfolio(strategy="agressiva")
+        pf_hodl = Portfolio(strategy="hodl")
+
+        results_history = []
+        for k in klines:
+            ts = datetime.fromtimestamp(k[0]/1000).strftime("%Y-%m-%d")
+            open_p, high_p, low_p, close_p = float(k[1]), float(k[2]), float(k[3]), float(k[4])
+            change_pct = ((close_p - open_p) / open_p) * 100
+
+            # Mock Brief Data baseado no kline real
+            sig = "ALTA" if change_pct > 0 else "BAIXA" if change_pct < 0 else "NEUTRO"
+            conf = min(95, max(65, int(70 + abs(change_pct) * 5)))
+            rsi = min(85, max(15, int(50 + change_pct * 5)))
+
+            brief_mock = {
+                "price": close_p,
+                "signal": sig,
+                "confidence": conf,
+                "rsi": rsi,
+                "net_signal": int(change_pct),
+                "summary": f"Backtest kline {ts}: Close ${close_p:,.2f} ({change_pct:+.2f}%)"
+            }
+
+            # Decio Decision (forçando local para rapidez e consistência do backtest)
+            decio = DecioAgent(brief_mock)
+            decio._decide_llm = lambda: None
+            decio_res = decio.run()
+
+            # PaperT Execution (Soberana, Agressiva e HODL)
+            papert = PapertAgent(decio_res)
+            papert._current_price = lambda: close_p
+            papert._write_log = lambda x: True  # silencia log físico no backtest
+            
+            exec_res = papert.run()
+            results_history.append({
+                "date": ts,
+                "price": close_p,
+                "change_pct": round(change_pct, 2),
+                "decio_action": decio_res["action"],
+                "decio_conf": decio_res["confidence"],
+                "soberana_bal": exec_res["portfolio"]["balance"],
+                "agressiva_bal": exec_res["portfolio"]["agressiva"]["balance"],
+                "hodl_bal": exec_res["portfolio"]["hodl"]["balance"]
+            })
+
+        # Coleta estatísticas finais do backtest
+        stats_final = pf_soberana.get_stats()
+        stats_agr = pf_agressiva.get_stats()
+        stats_hodl = pf_hodl.get_stats()
+
+        # Restaura variáveis originais
+        papert_module.PORTFOLIO_DB = old_db
+        papert_module.INITIAL_BALANCE = old_init
+
+        return jsonify({
+            "status": "completed",
+            "days_simulated": len(klines),
+            "initial_capital": initial_capital,
+            "final_stats": {
+                "soberana": {
+                    "balance": stats_final["current_balance"],
+                    "total_pnl": stats_final["total_pnl"],
+                    "total_pnl_pct": stats_final["total_pnl_pct"],
+                    "win_rate": stats_final["win_rate"],
+                    "max_drawdown": stats_final["max_drawdown"]
+                },
+                "agressiva": {
+                    "balance": stats_agr["current_balance"],
+                    "total_pnl": stats_agr["total_pnl"],
+                    "total_pnl_pct": stats_agr["total_pnl_pct"],
+                    "win_rate": stats_agr["win_rate"],
+                    "max_drawdown": stats_agr["max_drawdown"]
+                },
+                "hodl": {
+                    "balance": stats_hodl["current_balance"],
+                    "total_pnl": stats_hodl["total_pnl"],
+                    "total_pnl_pct": stats_hodl["total_pnl_pct"],
+                    "max_drawdown": stats_hodl["max_drawdown"]
+                }
+            },
+            "history": results_history
+        })
+    except Exception as e:
+        return jsonify({"error": str(e)}), 500
+
+# ── Config API (Cockpit de Dosagens de Trabalho) ──────────────────────────────
+CONFIG_FILE = os.path.join(DATA_DIR, "user_config.json")
+
+DEFAULT_CONFIG = {
+    "frequency": "1h",         # 4h, 1h, 15m
+    "risk_lock": 70,           # 75, 70, 60
+    "capital": 10000.0,
+    "max_allocation": 25,      # pct
+    "weight_balance": 50       # 0 (Tech) to 100 (Sentiment)
+}
+
+@app.route("/api/config", methods=["GET"])
+def get_config():
+    if not os.path.exists(CONFIG_FILE):
+        with open(CONFIG_FILE, "w") as f:
+            json.dump(DEFAULT_CONFIG, f)
+        return jsonify(DEFAULT_CONFIG)
+    try:
+        with open(CONFIG_FILE, "r") as f:
+            cfg = json.load(f)
+            for k, v in DEFAULT_CONFIG.items():
+                if k not in cfg: cfg[k] = v
+            return jsonify(cfg)
+    except:
+        return jsonify(DEFAULT_CONFIG)
+
+@app.route("/api/config", methods=["POST"])
+def save_config():
+    try:
+        data = request.json or {}
+        cfg = {
+            "frequency": data.get("frequency", "1h"),
+            "risk_lock": int(data.get("risk_lock", 70)),
+            "capital": float(data.get("capital", 10000.0)),
+            "max_allocation": int(data.get("max_allocation", 25)),
+            "weight_balance": int(data.get("weight_balance", 50))
+        }
+        with open(CONFIG_FILE, "w") as f:
+            json.dump(cfg, f, indent=2)
+            
+        # Atualiza o INITIAL_BALANCE do PaperT em tempo real se o capital mudar
+        sys.path.insert(0, "/app/agents")
+        try:
+            import agents.papert as papert_module
+            papert_module.INITIAL_BALANCE = cfg["capital"]
+        except: pass
+        
+        return jsonify({"status": "saved", "config": cfg})
+    except Exception as e:
+        return jsonify({"error": str(e)}), 400
+
+# ── Background Automated Scheduler ───────────────────────────────────────────
+def start_scheduler():
+    def scheduler_loop():
+        import time
+        # Sleep for a bit to let the app start up completely
+        time.sleep(15)
+        app.logger.info("Scheduler thread started successfully.")
+        while True:
+            try:
+                # Load configuration to get frequency
+                freq_mins = 60
+                if os.path.exists(CONFIG_FILE):
+                    try:
+                        with open(CONFIG_FILE, "r") as f:
+                            cfg = json.load(f)
+                            freq = cfg.get("frequency", "1h")
+                            if freq == "15m":
+                                freq_mins = 15
+                            elif freq == "4h":
+                                freq_mins = 240
+                            else:
+                                freq_mins = 60
+                    except:
+                        pass
+                
+                # Check last pipeline run from database
+                should_run = False
+                try:
+                    conn = _db_conn(DB_PATH)
+                    c = conn.cursor()
+                    row = c.execute("SELECT started_at FROM pipeline_runs ORDER BY started_at DESC LIMIT 1").fetchone()
+                    conn.close()
+                except Exception as dbe:
+                    row = None
+                    app.logger.error(f"Scheduler DB check failed: {dbe}")
+                
+                if not row:
+                    should_run = True
+                else:
+                    last_run_str = row[0]
+                    try:
+                        last_run = datetime.fromisoformat(last_run_str)
+                        if datetime.now() - last_run >= timedelta(minutes=freq_mins):
+                            should_run = True
+                    except Exception as pe:
+                        should_run = True
+                
+                if should_run:
+                    if not state.pipeline_running:
+                        app.logger.info(f"Scheduler triggering automated pipeline run (Interval: {freq_mins}m)...")
+                        thread = threading.Thread(target=run_pipeline)
+                        thread.start()
+                    else:
+                        app.logger.info("Scheduler skipped run: pipeline is already running.")
+            except Exception as e:
+                app.logger.error(f"Scheduler loop error: {e}")
+            
+            # Check every 60 seconds
+            time.sleep(60)
+
+    thread = threading.Thread(target=scheduler_loop, daemon=True)
+    thread.start()
+
+start_scheduler()
+
 if __name__ == "__main__":
-    app.run(host="0.0.0.0", port=3100, debug=False)
+    app.run(host="0.0.0.0", port=3100, debug=False)
+# ── DEBUG: Log all DB operations ─────────────────────────────────────────────