From 9b9926b8b823808e8c9a0c8f599295320944dedf Mon Sep 17 00:00:00 2001 From: Harikeshav Rameshkumar Date: Fri, 27 Feb 2026 21:25:57 -0500 Subject: [PATCH 1/2] feat(brain): introduce background scheduler with market sync and scan automation - Added `apscheduler` integration to enable automated background tasks within the Brain service. - Implemented a scheduler that periodically: - **Syncs markets** from the Gamma API every 15 minutes. - **Scans market pairs** for analysis using pre-defined logic. - Refactored market sync (`_internal_sync_markets`) and scan logic (`_internal_scan_markets`) into reusable functions for both API endpoints and background jobs. - Enhanced error handling and logging: - Added structured logs for sync and scan tasks via API and background jobs. - Improved exception handling to retain service stability. - Introduced shutdown hooks for graceful cleanup of the scheduler during service exit. - Updated dependencies: - Included `apscheduler>=3.11.2` in `pyproject.toml` and `uv.lock`. - Added `auto_discover_and_analyze` method for integrated sync and scan in the background. This update automates critical market update and analysis workflows, reducing manual overhead and improving operational efficiency. --- services/brain/main.py | 128 ++++++++++++++++++++++------------ services/brain/pyproject.toml | 1 + services/brain/uv.lock | 26 +++++++ 3 files changed, 110 insertions(+), 45 deletions(-) diff --git a/services/brain/main.py b/services/brain/main.py index c1c428c..bc3da51 100644 --- a/services/brain/main.py +++ b/services/brain/main.py @@ -19,6 +19,8 @@ from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession +from apscheduler.schedulers.asyncio import AsyncIOScheduler + from config import get_settings from database import close_db, get_session, init_db from gamma_client import GammaClient @@ -54,6 +56,7 @@ _graph_manager = RelationshipGraphManager() _gamma_client: GammaClient | None = None +_scheduler = AsyncIOScheduler() _api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False) @@ -103,10 +106,22 @@ async def lifespan(app: FastAPI): async for session in get_session(): await _graph_manager.load_from_db(session) + # Background tasks + _scheduler.add_job( + auto_discover_and_analyze, + "interval", + minutes=15, + id="auto_discover_job", + replace_existing=True, + ) + _scheduler.start() + logger.info("brain_scheduler_started") + logger.info("brain_ready") yield # Shutdown + _scheduler.shutdown() if _gamma_client: await _gamma_client.aclose() await close_db() @@ -149,11 +164,10 @@ async def health(session: AsyncSession = Depends(get_session)): # ── Market Sync ─────────────────────────────────────────────────────────────── -@app.post("/markets/sync", response_model=dict, dependencies=[Depends(_verify_admin)]) -async def sync_markets(session: AsyncSession = Depends(get_session)): - """Fetch active markets from the Gamma API and upsert into the local DB.""" +async def _internal_sync_markets(session: AsyncSession) -> dict[str, int]: + """Internal sync logic so background jobs can call it directly.""" if _gamma_client is None: - raise HTTPException(status_code=503, detail="Gamma client not initialised") + raise RuntimeError("Gamma client not initialised") settings = get_settings() raw_markets = await _gamma_client.fetch_all_active_markets( @@ -176,7 +190,6 @@ async def sync_markets(session: AsyncSession = Depends(get_session)): now = datetime.now(UTC) - # Extract token IDs from the market data clob_token_ids: list[str] = [] tokens = raw.get("tokens") or raw.get("clobTokenIds") if isinstance(tokens, list): @@ -216,14 +229,21 @@ async def sync_markets(session: AsyncSession = Depends(get_session)): created += 1 await session.commit() - logger.bind( - total_fetched=len(raw_markets), - created=created, - updated=updated, - ).info("markets_synced") return {"created": created, "updated": updated, "total_fetched": len(raw_markets)} +@app.post("/markets/sync", response_model=dict, dependencies=[Depends(_verify_admin)]) +async def sync_markets(session: AsyncSession = Depends(get_session)): + """Fetch active markets from the Gamma API and upsert into the local DB.""" + try: + stats = await _internal_sync_markets(session) + logger.bind(**stats).info("markets_synced_via_api") + return stats + except Exception as e: + logger.exception("api_sync_markets_failed") + raise HTTPException(status_code=500, detail=str(e)) + + @app.get("/markets", response_model=list[MarketRead]) async def list_markets( limit: int = Query(default=50, le=200), @@ -306,19 +326,10 @@ async def analyze_pair( # ── Scan ────────────────────────────────────────────────────────────────────── -@app.post("/scan", response_model=dict, dependencies=[Depends(_verify_admin)]) -async def scan_markets( - limit: int = Query(default=20, le=100, description="Max markets to compare"), - session: AsyncSession = Depends(get_session), -): - """Trigger a scan: fetch top markets and analyze unprocessed pairs. - - This is a heavy endpoint — it iterates over C(n,2) pairs and invokes - the LLM for each. Use ``limit`` to control blast radius. - """ +async def _internal_scan_markets(session: AsyncSession, limit: int = 20) -> dict[str, int]: + """Internal scan logic so background jobs can call it directly.""" settings = get_settings() - # Get top markets by volume result = await session.execute( select(Market) .where(Market.active.is_(True), Market.closed.is_(False)) @@ -327,15 +338,9 @@ async def scan_markets( ) markets = result.scalars().all() - logger.bind( - active_markets=len(markets), - admin=True, - ).info("scan_started") - if len(markets) < 2: - return {"analyzed": 0, "message": "Not enough markets to compare. Sync markets first."} + return {"analyzed": 0, "errors": 0, "total_markets": len(markets)} - # Pre-compile the graph once per request to avoid O(n^2) overhead compiled_graph = build_analysis_graph(settings, session) analyzed = 0 @@ -344,8 +349,6 @@ async def scan_markets( for i, market_a in enumerate(markets): for market_b in markets[i + 1:]: - # Skip already-analyzed pairs (even if inactive right now, unless we want - # to re-scan them) existing = await session.execute( select(Relationship).where( Relationship.is_active.is_(True), @@ -363,10 +366,6 @@ async def scan_markets( skipped += 1 continue - logger.bind( - a=market_a.condition_id, - b=market_b.condition_id, - ).info("scan_pair") try: state: AnalysisState = { "condition_id_a": market_a.condition_id, @@ -394,20 +393,59 @@ async def scan_markets( analyzed += 1 else: errors += 1 - logger.bind(error=graph_result["error"]).warning("scan_pair_error") except Exception: errors += 1 - logger.bind( - a=market_a.condition_id, - b=market_b.condition_id, - ).exception("scan_pair_exception") - logger.bind( - analyzed=analyzed, - skipped=skipped, - errors=errors, - ).info("scan_complete") - return {"analyzed": analyzed, "errors": errors, "total_markets": len(markets)} + return {"analyzed": analyzed, "errors": errors, + "total_markets": len(markets), "skipped": skipped} + + +@app.post("/scan", response_model=dict, dependencies=[Depends(_verify_admin)]) +async def scan_markets( + limit: int = Query(default=20, le=100, description="Max markets to compare"), + session: AsyncSession = Depends(get_session), +): + """Trigger a scan: fetch top markets and analyze unprocessed pairs. + + This is a heavy endpoint — it iterates over C(n,2) pairs and invokes + the LLM for each. Use ``limit`` to control blast radius. + """ + try: + logger.bind(admin=True).info("scan_markets_api_start") + stats = await _internal_scan_markets(session, limit) + logger.bind(**stats).info("scan_markets_api_complete") + return stats + except Exception as e: + logger.exception("api_scan_markets_failed") + raise HTTPException(status_code=500, detail=str(e)) + + +async def auto_discover_and_analyze() -> None: + """Background task to sync latest markets and scan pairs.""" + logger.info("background_sync_and_scan_started") + try: + # We need a fresh session for the background task. + # get_session yields one, so we can use async for or anext + session_gen = get_session() + session = await anext(session_gen) + try: + # 1. Sync markets from API + sync_stats = await _internal_sync_markets(session) + + # 2. Re-evaluate a small batch of pairs (limit=20 for speed) + scan_stats = await _internal_scan_markets(session, limit=20) + + logger.bind( + sync=sync_stats, + scan=scan_stats + ).info("background_sync_and_scan_completed") + finally: + try: + await anext(session_gen) + except StopAsyncIteration: + pass + except Exception: + logger.exception("background_sync_and_scan_failed") # ── Relationships ───────────────────────────────────────────────────────────── diff --git a/services/brain/pyproject.toml b/services/brain/pyproject.toml index e1196e0..d858e0c 100644 --- a/services/brain/pyproject.toml +++ b/services/brain/pyproject.toml @@ -16,6 +16,7 @@ dependencies = [ "httpx>=0.28.0", "loguru>=0.7.0", "networkx>=3.4", + "apscheduler>=3.11.2", ] [project.optional-dependencies] diff --git a/services/brain/uv.lock b/services/brain/uv.lock index 2b87930..d5903c8 100644 --- a/services/brain/uv.lock +++ b/services/brain/uv.lock @@ -32,11 +32,24 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/38/0e/27be9fdef66e72d64c0cdc3cc2823101b80585f8119b5c112c2e8f5f7dab/anyio-4.12.1-py3-none-any.whl", hash = "sha256:d405828884fc140aa80a3c667b8beed277f1dfedec42ba031bd6ac3db606ab6c", size = 113592, upload-time = "2026-01-06T11:45:19.497Z" }, ] +[[package]] +name = "apscheduler" +version = "3.11.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "tzlocal" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/07/12/3e4389e5920b4c1763390c6d371162f3784f86f85cd6d6c1bfe68eef14e2/apscheduler-3.11.2.tar.gz", hash = "sha256:2a9966b052ec805f020c8c4c3ae6e6a06e24b1bf19f2e11d91d8cca0473eef41", size = 108683, upload-time = "2025-12-22T00:39:34.884Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/9f/64/2e54428beba8d9992aa478bb8f6de9e4ecaa5f8f513bcfd567ed7fb0262d/apscheduler-3.11.2-py3-none-any.whl", hash = "sha256:ce005177f741409db4e4dd40a7431b76feb856b9dd69d57e0da49d6715bfd26d", size = 64439, upload-time = "2025-12-22T00:39:33.303Z" }, +] + [[package]] name = "arbos-brain" version = "0.1.0" source = { virtual = "." } dependencies = [ + { name = "apscheduler" }, { name = "asyncpg" }, { name = "fastapi" }, { name = "httpx" }, @@ -61,6 +74,7 @@ dev = [ [package.metadata] requires-dist = [ + { name = "apscheduler", specifier = ">=3.11.2" }, { name = "asyncpg", specifier = ">=0.30.0" }, { name = "fastapi", specifier = ">=0.115.0" }, { name = "httpx", specifier = ">=0.28.0" }, @@ -1088,6 +1102,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c7/b0/003792df09decd6849a5e39c28b513c06e84436a54440380862b5aeff25d/tzdata-2025.3-py2.py3-none-any.whl", hash = "sha256:06a47e5700f3081aab02b2e513160914ff0694bce9947d6b76ebd6bf57cfc5d1", size = 348521, upload-time = "2025-12-13T17:45:33.889Z" }, ] +[[package]] +name = "tzlocal" +version = "5.3.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "tzdata", marker = "sys_platform == 'win32'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/8b/2e/c14812d3d4d9cd1773c6be938f89e5735a1f11a9f184ac3639b93cef35d5/tzlocal-5.3.1.tar.gz", hash = "sha256:cceffc7edecefea1f595541dbd6e990cb1ea3d19bf01b2809f362a03dd7921fd", size = 30761, upload-time = "2025-03-05T21:17:41.549Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c2/14/e2a54fabd4f08cd7af1c07030603c3356b74da07f7cc056e600436edfa17/tzlocal-5.3.1-py3-none-any.whl", hash = "sha256:eb1a66c3ef5847adf7a834f1be0800581b683b5608e74f86ecbcef8ab91bb85d", size = 18026, upload-time = "2025-03-05T21:17:39.857Z" }, +] + [[package]] name = "urllib3" version = "2.6.3" From 00a517443dda1ebf064a3345868503688075a13b Mon Sep 17 00:00:00 2001 From: Harikeshav Rameshkumar Date: Fri, 27 Feb 2026 21:37:16 -0500 Subject: [PATCH 2/2] feat(brain): enhance scheduler job configuration, error handling, and analysis response messaging - Enhanced `auto_discover_job` scheduler: - Added `max_instances=1` to prevent overlapping job executions. - Enabled `coalesce` to merge missed runs during downtime. - Set `misfire_grace_time=300` to allow a 5-minute buffer for delayed jobs. - Improved error handling for API endpoints: - Updated generic exceptions to return a standardized "Internal server error" message for better consistency and user experience. - Explicitly re-raised `HTTPException` without modification. - Refined response structure for market analysis: - Added more descriptive messages for scenarios with insufficient markets or analyzed pairs. - Included `skipped` field to clarify skipped market counts in analysis results. These updates improve scheduler reliability, API robustness, and response clarity in the Brain service. --- services/brain/main.py | 43 ++++++++++++++++++++++++++---------------- 1 file changed, 27 insertions(+), 16 deletions(-) diff --git a/services/brain/main.py b/services/brain/main.py index bc3da51..e054468 100644 --- a/services/brain/main.py +++ b/services/brain/main.py @@ -113,6 +113,9 @@ async def lifespan(app: FastAPI): minutes=15, id="auto_discover_job", replace_existing=True, + max_instances=1, + coalesce=True, + misfire_grace_time=300, ) _scheduler.start() logger.info("brain_scheduler_started") @@ -239,9 +242,11 @@ async def sync_markets(session: AsyncSession = Depends(get_session)): stats = await _internal_sync_markets(session) logger.bind(**stats).info("markets_synced_via_api") return stats - except Exception as e: + except HTTPException: + raise + except Exception: logger.exception("api_sync_markets_failed") - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail="Internal server error") @app.get("/markets", response_model=list[MarketRead]) @@ -339,7 +344,13 @@ async def _internal_scan_markets(session: AsyncSession, limit: int = 20) -> dict markets = result.scalars().all() if len(markets) < 2: - return {"analyzed": 0, "errors": 0, "total_markets": len(markets)} + return { + "message": "Not enough markets to analyze", + "analyzed": 0, + "errors": 0, + "total_markets": len(markets), + "skipped": 0, + } compiled_graph = build_analysis_graph(settings, session) @@ -396,8 +407,13 @@ async def _internal_scan_markets(session: AsyncSession, limit: int = 20) -> dict except Exception: errors += 1 - return {"analyzed": analyzed, "errors": errors, - "total_markets": len(markets), "skipped": skipped} + return { + "message": f"Analyzed {analyzed} pairs, skipped {skipped}, {errors} errors", + "analyzed": analyzed, + "errors": errors, + "total_markets": len(markets), + "skipped": skipped, + } @app.post("/scan", response_model=dict, dependencies=[Depends(_verify_admin)]) @@ -415,9 +431,11 @@ async def scan_markets( stats = await _internal_scan_markets(session, limit) logger.bind(**stats).info("scan_markets_api_complete") return stats - except Exception as e: + except HTTPException: + raise + except Exception: logger.exception("api_scan_markets_failed") - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail="Internal server error") async def auto_discover_and_analyze() -> None: @@ -425,10 +443,7 @@ async def auto_discover_and_analyze() -> None: logger.info("background_sync_and_scan_started") try: # We need a fresh session for the background task. - # get_session yields one, so we can use async for or anext - session_gen = get_session() - session = await anext(session_gen) - try: + async for session in get_session(): # 1. Sync markets from API sync_stats = await _internal_sync_markets(session) @@ -439,11 +454,7 @@ async def auto_discover_and_analyze() -> None: sync=sync_stats, scan=scan_stats ).info("background_sync_and_scan_completed") - finally: - try: - await anext(session_gen) - except StopAsyncIteration: - pass + break except Exception: logger.exception("background_sync_and_scan_failed")