Update app_routes.py

app_routes.py

@@ -1,6 +1,7 @@
 from fastapi import APIRouter, Request
 from fastapi.responses import StreamingResponse
-import aiosqlite, asyncio, random, json, logging
+import aiosqlite, asyncio, random, json, logging, time
+from collections import defaultdict
 from npc_core import GroqAIClient, get_db, get_db_read
 from npc_trading import (
     ALL_TICKERS, TRADING_STRATEGIES, IDENTITY_STRATEGY_MAP,
@@ -21,12 +22,61 @@ router = APIRouter()
 _DB_PATH = ""
 _GROQ_KEY = ""
 _EventBus = None
+
+# ══════════════════════════════════════════════════
+# ① In-Memory Cache — Republic dashboard, prices
+# ══════════════════════════════════════════════════
+class _TimedCache:
+    """Simple TTL cache for expensive queries"""
+    def __init__(self):
+        self._store = {}
+        self._lock = asyncio.Lock()
+    async def get(self, key: str, ttl: float, factory):
+        now = time.time()
+        if key in self._store:
+            val, ts = self._store[key]
+            if now - ts < ttl:
+                return val
+        async with self._lock:
+            # Double-check after lock acquisition
+            if key in self._store:
+                val, ts = self._store[key]
+                if now - ts < ttl:
+                    return val
+            val = await factory()
+            self._store[key] = (val, time.time())
+            return val
+    def invalidate(self, key: str = None):
+        if key:
+            self._store.pop(key, None)
+        else:
+            self._store.clear()
+
+_cache = _TimedCache()
+
+# ══════════════════════════════════════════════════
+# ⑨ Rate Limiter — POST endpoint protection
+# ══════════════════════════════════════════════════
+class _RateLimiter:
+    """Simple in-memory rate limiter: max N requests per window per key"""
+    def __init__(self):
+        self._hits = defaultdict(list)
+    def check(self, key: str, max_requests: int = 10, window_sec: int = 60) -> bool:
+        now = time.time()
+        cutoff = now - window_sec
+        self._hits[key] = [t for t in self._hits[key] if t > cutoff]
+        if len(self._hits[key]) >= max_requests:
+            return False  # rate limited
+        self._hits[key].append(now)
+        return True
+
+_limiter = _RateLimiter()
 def configure(db_path: str, groq_key: str, event_bus_cls):
     global _DB_PATH, _GROQ_KEY, _EventBus
     _DB_PATH, _GROQ_KEY, _EventBus = db_path, groq_key, event_bus_cls
 @router.get("/api/trading/prices")
 async def api_trading_prices():
-    return await get_all_prices(_DB_PATH)
+    return await _cache.get('trading_prices', 60.0, lambda: get_all_prices(_DB_PATH))
 @router.get("/api/trading/ticker/{ticker}")
 async def api_ticker_detail(ticker: str):
     return await get_ticker_positions(_DB_PATH, ticker)
@@ -67,8 +117,7 @@ async def api_market_indices():
     return {"indices": indices}
 @router.get("/api/intelligence/screening/{ticker}")
 async def api_screening_data(ticker: str):
-    async with get_db(_DB_PATH) as db:
-        await db.execute("PRAGMA busy_timeout=30000")
+    async with get_db_read(_DB_PATH) as db:
         try:
             cursor = await db.execute("SELECT ticker, price, change_pct, volume, market_cap, rsi, pe_ratio, high_52w, low_52w, from_high, from_low FROM market_prices WHERE ticker=?", (ticker,))
             row = await cursor.fetchone()
@@ -77,8 +126,7 @@ async def api_screening_data(ticker: str):
     return {"ticker": ticker, "error": "no data"}
 @router.get("/api/intelligence/target/{ticker}")
 async def api_target_price(ticker: str):
-    async with get_db(_DB_PATH) as db:
-        await db.execute("PRAGMA busy_timeout=30000")
+    async with get_db_read(_DB_PATH) as db:
         try:
             cursor = await db.execute("SELECT price, rsi, pe_ratio, from_high, from_low FROM market_prices WHERE ticker=?", (ticker,))
             row = await cursor.fetchone()
@@ -94,8 +142,7 @@ async def api_target_price(ticker: str):
 async def api_news_feed(ticker: str = None, limit: int = 50):
     news = await load_news_from_db(_DB_PATH, ticker, limit)
     try:
-        async with get_db(_DB_PATH) as db:
-            await db.execute("PRAGMA busy_timeout=30000")
+        async with get_db_read(_DB_PATH) as db:
             for n in news:
                 nid = n.get('id')
                 if nid:
@@ -108,6 +155,7 @@ async def api_news_feed(ticker: str = None, limit: int = 50):
 async def api_news_react(request: Request):
     body = await request.json(); email = body.get("email", ""); news_id = body.get("news_id"); emoji = body.get("emoji", "")
     if not email or not news_id or emoji not in ('👍','🔥','😱','💀','🚀','😂'): return {"error": "Invalid reaction"}
+    if not _limiter.check(f'react:{email}', 30, 60): return {"error": "Rate limited"}
     try:
         async with get_db(_DB_PATH) as db:
             await db.execute("PRAGMA busy_timeout=30000"); await db.execute("INSERT OR IGNORE INTO news_reactions (news_id, user_email, emoji) VALUES (?,?,?)", (news_id, email, emoji)); await db.commit()
@@ -117,11 +165,14 @@ async def api_news_react(request: Request):
 @router.get("/api/market/pulse")
 async def api_market_pulse():
     try:
-        pulse = await get_market_pulse(_DB_PATH); pulse['indices'] = await load_indices_from_db(_DB_PATH) or []
-        return pulse
+        return await _cache.get('market_pulse', 45.0, _market_pulse_impl)
     except Exception as e:
         logger.error(f"Market pulse error: {e}")
         return {"hot_movers": [], "indices": [], "activity": {}}
+
+async def _market_pulse_impl():
+    pulse = await get_market_pulse(_DB_PATH); pulse['indices'] = await load_indices_from_db(_DB_PATH) or []
+    return pulse
 @router.get("/api/research/feed")
 async def api_research_feed(ticker: str = None, limit: int = 30):
     reports = await get_research_feed(_DB_PATH, ticker, min(limit, 50)); stats = await get_research_stats(_DB_PATH)
@@ -150,8 +201,8 @@ async def api_deep_analysis(ticker: str):
 @router.get("/api/analysis/all/summary")
 async def api_all_analyses():
     analyses = await load_all_analyses_from_db(_DB_PATH)
-    async with get_db(_DB_PATH) as db:
-        await db.execute("PRAGMA busy_timeout=30000"); col_cursor = await db.execute("PRAGMA table_info(market_prices)")
+    async with get_db_read(_DB_PATH) as db:
+        col_cursor = await db.execute("PRAGMA table_info(market_prices)")
         existing_cols = {r[1] for r in await col_cursor.fetchall()}
         has_rsi = 'rsi' in existing_cols; has_pe = 'pe_ratio' in existing_cols; has_52w = 'high_52w' in existing_cols; has_from = 'from_high' in existing_cols
         base_cols = "ticker, price, change_pct, prev_close, volume, high_24h, low_24h, market_cap"; extra_cols = ""
@@ -219,14 +270,13 @@ async def api_all_analyses():
 @router.get("/api/npc/search")
 async def api_npc_search(q: str = ""):
     if len(q) < 1: return {"npcs": []}
-    async with get_db(_DB_PATH) as db:
-        await db.execute("PRAGMA busy_timeout=30000"); cursor = await db.execute("SELECT agent_id, username, ai_identity FROM npc_agents WHERE username LIKE ? AND is_active=1 LIMIT 10", (f'%{q}%',))
+    async with get_db_read(_DB_PATH) as db:
+        cursor = await db.execute("SELECT agent_id, username, ai_identity FROM npc_agents WHERE username LIKE ? AND is_active=1 LIMIT 10", (f'%{q}%',))
         return {"npcs": [{"agent_id": r[0], "username": r[1], "identity": r[2]} for r in await cursor.fetchall()]}
 @router.get("/api/npc/profile/{agent_id}")
 async def api_npc_profile(agent_id: str):
     try:
-        async with get_db(_DB_PATH) as db:
-            await db.execute("PRAGMA busy_timeout=30000")
+        async with get_db_read(_DB_PATH) as db:
             cursor = await db.execute("SELECT agent_id, username, ai_identity, mbti, gpu_dollars, created_at FROM npc_agents WHERE agent_id=?", (agent_id,))
             row = await cursor.fetchone()
             if not row: return {"error": "NPC not found"}
@@ -302,8 +352,7 @@ async def api_sec_dashboard():
         return {"stats": {"total_violations": 0, "total_fines_gpu": 0, "active_suspensions": 0, "pending_reports": 0}, "announcements": [], "top_violators": [], "recent_reports": []}
 @router.get("/api/sec/violations/{agent_id}")
 async def api_sec_violations(agent_id: str):
-    async with get_db(_DB_PATH) as db:
-        await db.execute("PRAGMA busy_timeout=30000")
+    async with get_db_read(_DB_PATH) as db:
         cursor = await db.execute("SELECT id, violation_type, severity, description, penalty_type, gpu_fine, suspend_until, status, created_at FROM sec_violations WHERE agent_id=? ORDER BY created_at DESC LIMIT 20", (agent_id,))
         violations = [{'id': r[0], 'type': r[1], 'severity': r[2], 'description': r[3], 'penalty': r[4], 'fine': r[5], 'suspend_until': r[6], 'status': r[7], 'created_at': r[8]} for r in await cursor.fetchall()]
         cursor2 = await db.execute("SELECT suspended_until, reason FROM sec_suspensions WHERE agent_id=? AND suspended_until > datetime('now')", (agent_id,))
@@ -352,14 +401,12 @@ async def api_sec_seed():
     except Exception as e: return {"error": str(e)}
 @router.get("/api/sec/announcements")
 async def api_sec_announcements(limit: int = 30):
-    async with get_db(_DB_PATH) as db:
-        await db.execute("PRAGMA busy_timeout=30000")
+    async with get_db_read(_DB_PATH) as db:
         cursor = await db.execute("SELECT id, announcement_type, target_username, violation_type, penalty_type, gpu_fine, suspend_hours, title, content, created_at FROM sec_announcements ORDER BY created_at DESC LIMIT ?", (min(limit, 50),))
         return {"announcements": [{'id': r[0], 'type': r[1], 'target': r[2], 'violation': r[3], 'penalty': r[4], 'fine': r[5], 'hours': r[6], 'title': r[7], 'content': r[8], 'created_at': r[9]} for r in await cursor.fetchall()]}
 @router.get("/api/sec/suspended")
 async def api_sec_suspended():
-    async with get_db(_DB_PATH) as db:
-        await db.execute("PRAGMA busy_timeout=30000")
+    async with get_db_read(_DB_PATH) as db:
         cursor = await db.execute("SELECT s.agent_id, n.username, s.reason, s.suspended_until, s.created_at FROM sec_suspensions s JOIN npc_agents n ON s.agent_id = n.agent_id WHERE s.suspended_until > datetime('now') ORDER BY s.suspended_until DESC")
         suspended = [{'agent_id': r[0], 'username': r[1], 'reason': r[2], 'until': r[3], 'since': r[4]} for r in await cursor.fetchall()]
     return {"suspended": suspended, "count": len(suspended)}
@@ -369,6 +416,7 @@ async def api_tip_npc(request: Request):
     user_email = body.get("email", ""); target_agent = body.get("target_agent_id", ""); amount = int(body.get("amount", 0)); message = body.get("message", "").strip()[:200]
     if not user_email or not target_agent or amount < 10: return {"error": "email, target_agent_id, amount(>=10) required"}
     if amount > 1000: return {"error": "Max tip: 1,000 GPU"}
+    if not _limiter.check(f'tip:{user_email}', 10, 60): return {"error": "Rate limited — max 10 tips per minute"}
     async with get_db(_DB_PATH) as db:
         await db.execute("PRAGMA busy_timeout=30000"); cursor = await db.execute("SELECT gpu_dollars, username FROM user_profiles WHERE email=?", (user_email,))
         user = await cursor.fetchone()
@@ -392,6 +440,7 @@ async def api_influence_npc(request: Request):
     body = await request.json()
     user_email = body.get("email", ""); target_agent = body.get("target_agent_id", ""); ticker = body.get("ticker", "").upper(); stance = body.get("stance", "").lower()
     if not all([user_email, target_agent, ticker, stance]) or stance not in ('bullish', 'bearish'): return {"error": "email, target_agent_id, ticker, stance(bullish/bearish) required"}
+    if not _limiter.check(f'influence:{user_email}', 5, 60): return {"error": "Rate limited — max 5 influence attempts per minute"}
     async with get_db(_DB_PATH) as db:
         await db.execute("PRAGMA busy_timeout=30000"); cursor = await db.execute("SELECT username FROM user_profiles WHERE email=?", (user_email,))
         user = await cursor.fetchone(); cursor2 = await db.execute("SELECT username, ai_identity FROM npc_agents WHERE agent_id=?", (target_agent,))
@@ -415,8 +464,7 @@ async def api_influence_npc(request: Request):
     return {"status": "success", "influenced": influenced, "message": response_msg}
 @router.get("/api/swarm/trending")
 async def api_swarm_trending():
-    async with get_db(_DB_PATH) as db:
-        await db.execute("PRAGMA busy_timeout=30000")
+    async with get_db_read(_DB_PATH) as db:
         cursor = await db.execute("SELECT p.title || ' ' || p.content FROM posts p WHERE p.created_at > datetime('now', '-6 hours') ORDER BY p.likes_count DESC LIMIT 50")
         posts = [r[0] for r in await cursor.fetchall()]
     mentions = {}
@@ -430,8 +478,7 @@ async def api_swarm_trending():
 @router.get("/api/chat/messages")
 async def api_chat_messages(limit: int = 50, after_id: int = 0):
     try:
-        async with get_db(_DB_PATH) as db:
-            await db.execute("PRAGMA busy_timeout=30000")
+        async with get_db_read(_DB_PATH) as db:
             if after_id > 0:
                 cursor = await db.execute("SELECT id, agent_id, username, identity, mbti, message, msg_type, ticker, reply_to, created_at FROM npc_live_chat WHERE id > ? ORDER BY id ASC LIMIT ?", (after_id, limit))
             else:
@@ -443,8 +490,8 @@ async def api_chat_messages(limit: int = 50, after_id: int = 0):
 @router.get("/api/chat/stats")
 async def api_chat_stats():
     try:
-        async with get_db(_DB_PATH) as db:
-            await db.execute("PRAGMA busy_timeout=30000"); c1 = await db.execute("SELECT COUNT(*) FROM npc_live_chat"); total = (await c1.fetchone())[0]
+        async with get_db_read(_DB_PATH) as db:
+            c1 = await db.execute("SELECT COUNT(*) FROM npc_live_chat"); total = (await c1.fetchone())[0]
             c2 = await db.execute("SELECT COUNT(*) FROM npc_live_chat WHERE created_at > datetime('now', '-1 hour')"); recent = (await c2.fetchone())[0]
             c3 = await db.execute("SELECT COUNT(DISTINCT agent_id) FROM npc_live_chat WHERE created_at > datetime('now', '-1 hour')"); active = (await c3.fetchone())[0]
             return {'success': True, 'total': total, 'recent_1h': recent, 'active_chatters': active}
@@ -455,6 +502,7 @@ async def api_chat_send(request: Request):
         body = await request.json(); email = body.get("email", ""); message = body.get("message", "").strip()
         if not email or not message: return {"success": False, "error": "Email and message required"}
         if len(message) > 500: message = message[:500]
+        if not _limiter.check(f'chat:{email}', 10, 60): return {"success": False, "error": "Rate limited — max 10 messages per minute"}
         async with get_db(_DB_PATH) as db:
             await db.execute("PRAGMA busy_timeout=30000"); cursor = await db.execute("SELECT username FROM user_profiles WHERE email=?", (email,))
             row = await cursor.fetchone()
@@ -590,8 +638,7 @@ async def api_live_news(hours: int = 24):
     breaking = []
     
     try:
-        async with get_db(_DB_PATH) as db:
-            await db.execute("PRAGMA busy_timeout=30000")
+        async with get_db_read(_DB_PATH) as db:
             
             # ===== 1. LIQUIDATIONS (biggest drama) =====
             try:
@@ -954,10 +1001,13 @@ def _calc_happiness(win_rate, gpu_change_pct, social_score, loss_streak, sec_cou
 
 @router.get("/api/republic/dashboard")
 async def api_republic_dashboard():
+    """Cached republic dashboard — 60s TTL to avoid 25 heavy queries per request"""
+    return await _cache.get('republic_dashboard', 60.0, _republic_dashboard_impl)
+
+async def _republic_dashboard_impl():
     result = {}
     try:
-        async with get_db(_DB_PATH) as db:
-            await db.execute("PRAGMA busy_timeout=30000")
+        async with get_db_read(_DB_PATH) as db:
 
             # === POPULATION ===
             pop = {}
@@ -1449,6 +1499,7 @@ async def execute_random_event(db_path: str) -> dict:
     except Exception as e:
         logger.error(f"Random event execution error ({selected_key}): {e}")
 
+    _cache.invalidate('republic_dashboard')  # Invalidate dashboard cache after economic event
     return {
         'key': selected_key, 'name': ev['name'], 'emoji': ev['emoji'],
         'rarity': ev['rarity'], 'type': ev['type'],
@@ -1529,7 +1580,7 @@ async def check_npc_deaths(db_path: str) -> list:
 @router.get("/api/republic/events")
 async def api_republic_events():
     try:
-        async with get_db(_DB_PATH) as db:
+        async with get_db_read(_DB_PATH) as db:
             c = await db.execute("SELECT event_key, event_name, event_emoji, rarity, description, effect_summary, affected_count, gpu_impact, created_at FROM random_events ORDER BY created_at DESC LIMIT 20")
             events = [{'key': r[0], 'name': r[1], 'emoji': r[2], 'rarity': r[3], 'desc': r[4],
                        'effect': r[5], 'affected': r[6], 'gpu_impact': r[7], 'time': r[8]} for r in await c.fetchall()]
@@ -1540,7 +1591,7 @@ async def api_republic_events():
 @router.get("/api/republic/deaths")
 async def api_republic_deaths():
     try:
-        async with get_db(_DB_PATH) as db:
+        async with get_db_read(_DB_PATH) as db:
             c = await db.execute("""SELECT id, agent_id, username, ai_identity, mbti, cause, final_gpu, peak_gpu,
                 total_trades, lifespan_days, last_words, eulogy, resurrection_votes, resurrection_gpu, resurrected, created_at
                 FROM npc_deaths ORDER BY created_at DESC LIMIT 30""")
@@ -1574,6 +1625,7 @@ async def api_resurrect_npc(req: Request):
     amount = int(data.get('amount', 100))
     if not death_id or not donor_email: return {"error": "Missing death_id or email"}
     if amount < 10 or amount > 5000: return {"error": "Amount must be 10-5,000 GPU"}
+    if not _limiter.check(f'resurrect:{donor_email}', 5, 60): return {"error": "Rate limited — try again later"}
     try:
         async with get_db(_DB_PATH, write=True) as db:
             # Check donor balance
@@ -1720,6 +1772,41 @@ async def election_tick(db_path: str) -> dict:
                     used_identities.add(identity)
                     used_policies.add(best_policy)
 
+                # ④ Minimum 2 candidates — relax criteria if not enough
+                if len(candidates) < 2:
+                    logger.warning(f"🗳️ Only {len(candidates)} candidates from primary pool — relaxing GPU threshold")
+                    c_backup = await db.execute("""
+                        SELECT a.agent_id, a.username, a.ai_identity, a.mbti, a.gpu_dollars,
+                               COALESCE(a.total_likes_received,0) + COALESCE(a.post_count,0) as influence
+                        FROM npc_agents a WHERE a.is_active=1 AND a.gpu_dollars >= 1000
+                        AND a.agent_id NOT LIKE 'SEC_%'
+                        AND a.agent_id NOT IN (SELECT agent_id FROM election_candidates WHERE election_id=?)
+                        ORDER BY RANDOM() LIMIT 10
+                    """, (elec_id,))
+                    backup_pool = await c_backup.fetchall()
+                    for npc in backup_pool:
+                        if len(candidates) >= 2: break
+                        identity = npc[2]
+                        best_policy = None
+                        for pk in ELECTION_POLICIES:
+                            if pk not in used_policies:
+                                best_policy = pk; break
+                        if not best_policy: continue
+                        policy = ELECTION_POLICIES[best_policy]
+                        slogan = _random.choice(policy['slogan_pool'])
+                        await db.execute("""INSERT INTO election_candidates
+                            (election_id, agent_id, username, ai_identity, mbti, gpu_dollars, policy_key, policy_name, policy_desc, campaign_slogan)
+                            VALUES (?,?,?,?,?,?,?,?,?,?)""",
+                            (elec_id, npc[0], npc[1], npc[2], npc[3], npc[4], best_policy, policy['name'], policy['desc'], slogan))
+                        candidates.append({'username': npc[1], 'identity': npc[2], 'policy': policy['name']})
+                        used_policies.add(best_policy)
+                    if len(candidates) < 2:
+                        # Still not enough — cancel this election, retry next cycle
+                        await db.execute("UPDATE elections SET status='concluded' WHERE id=?", (elec_id,))
+                        await db.commit()
+                        logger.warning("🗳️ Election cancelled — insufficient candidates")
+                        return None
+
                 await db.commit()
                 return {'event': 'election_started', 'detail': f'{len(candidates)} candidates registered',
                         'candidates': candidates, 'election_id': elec_id}
@@ -1730,6 +1817,7 @@ async def election_tick(db_path: str) -> dict:
                 if datetime.utcnow() >= voting_start:
                     await db.execute("UPDATE elections SET status='voting' WHERE id=?", (current[0],))
                     await db.commit()
+                    _cache.invalidate('republic_dashboard')
                     return {'event': 'voting_started', 'detail': 'Polls are now open!', 'election_id': current[0]}
 
             elif current[1] == 'voting':
@@ -1795,7 +1883,7 @@ async def election_tick(db_path: str) -> dict:
 async def api_election_status():
     """Get current election status"""
     try:
-        async with get_db(_DB_PATH) as db:
+        async with get_db_read(_DB_PATH) as db:
             c = await db.execute("SELECT id, status, started_at, voting_starts_at, ends_at, winner_agent_id, winner_policy_key, total_votes, voter_turnout_pct FROM elections ORDER BY id DESC LIMIT 1")
             elec = await c.fetchone()
             if not elec: return {'status': 'no_election', 'message': 'No elections have been held yet'}
@@ -1848,6 +1936,7 @@ async def api_user_vote(req: Request):
     data = await req.json()
     email = data.get('email'); candidate_id = data.get('candidate_id')
     if not email or not candidate_id: return {"error": "Missing email or candidate_id"}
+    if not _limiter.check(f'vote:{email}', 3, 60): return {"error": "Rate limited — try again later"}
     try:
         async with get_db(_DB_PATH, write=True) as db:
             c = await db.execute("SELECT id FROM elections WHERE status='voting' ORDER BY id DESC LIMIT 1")
@@ -1881,8 +1970,7 @@ async def api_sse_stream():
     return StreamingResponse(event_generator(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})
 @router.get("/api/events/recent")
 async def api_recent_events(limit: int = 20):
-    async with get_db(_DB_PATH) as db:
-        await db.execute("PRAGMA busy_timeout=30000")
+    async with get_db_read(_DB_PATH) as db:
         cursor = await db.execute("SELECT p.agent_id, n.username, n.ai_identity, p.ticker, p.direction, p.gpu_bet, p.leverage, p.opened_at FROM npc_positions p JOIN npc_agents n ON p.agent_id = n.agent_id WHERE p.opened_at > datetime('now', '-1 hour') ORDER BY p.opened_at DESC LIMIT ?", (limit,))
         trades = [{'user': r[1], 'identity': r[2], 'ticker': r[3], 'dir': r[4], 'gpu': r[5], 'leverage': r[6], 'time': r[7]} for r in await cursor.fetchall()]
         cursor2 = await db.execute("SELECT p.agent_id, n.username, p.ticker, p.direction, p.gpu_bet, p.leverage, p.profit_gpu, p.profit_pct, p.liquidated, p.closed_at FROM npc_positions p JOIN npc_agents n ON p.agent_id = n.agent_id WHERE p.status IN ('closed', 'liquidated') AND p.closed_at > datetime('now', '-1 hour') ORDER BY p.closed_at DESC LIMIT ?", (limit,))