diff --git a/apps/osint-dashboard/.gitignore b/apps/osint-dashboard/.gitignore deleted file mode 100644 index c18dd8d..0000000 --- a/apps/osint-dashboard/.gitignore +++ /dev/null @@ -1 +0,0 @@ -__pycache__/ diff --git a/apps/osint-dashboard/Dockerfile b/apps/osint-dashboard/Dockerfile deleted file mode 100644 index 780d19b..0000000 --- a/apps/osint-dashboard/Dockerfile +++ /dev/null @@ -1,18 +0,0 @@ -FROM python:3.13-slim AS base - -WORKDIR /app - -RUN apt-get update && apt-get install -y --no-install-recommends \ - gcc libpq-dev \ - && rm -rf /var/lib/apt/lists/* - -COPY app/requirements.txt . -RUN pip install --no-cache-dir -r requirements.txt - -COPY app/ ./app/ -COPY alembic.ini ./alembic.ini -COPY alembic/ ./alembic/ - -EXPOSE 8000 -ENV PYTHONPATH=/app -CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/apps/osint-dashboard/alembic.ini b/apps/osint-dashboard/alembic.ini deleted file mode 100644 index a470ca8..0000000 --- a/apps/osint-dashboard/alembic.ini +++ /dev/null @@ -1,37 +0,0 @@ -[alembic] -script_location = alembic -# Override at runtime via DATABASE_URL environment variable (set in ConfigMap/Deployment) -sqlalchemy.url = driver://user:pass@localhost/dbname - -[loggers] -keys = root,sqlalchemy,alembic - -[handlers] -keys = console - -[formatters] -keys = generic - -[logger_root] -level = WARN -handlers = console - -[logger_sqlalchemy] -level = WARN -handlers = -qualname = sqlalchemy.engine - -[logger_alembic] -level = INFO -handlers = -qualname = alembic - -[handler_console] -class = StreamHandler -args = (sys.stderr,) -level = NOTSET -formatter = generic - -[formatter_generic] -format = %(levelname)-5.5s [%(name)s] %(message)s -datefmt = %H:%M:%S diff --git a/apps/osint-dashboard/alembic/env.py b/apps/osint-dashboard/alembic/env.py deleted file mode 100644 index 9b2fb7e..0000000 --- a/apps/osint-dashboard/alembic/env.py +++ /dev/null @@ -1,55 +0,0 @@ -import sys -from logging.config import fileConfig -from pathlib import Path - -from sqlalchemy import pool -from sqlalchemy.ext.asyncio import async_engine_from_config - -from alembic import context - -# Add app directory to path so we can import models/database -sys.path.insert(0, str(Path(__file__).parent.parent / "app")) - -from database import metadata, DATABASE_URL - -config = context.config -if config.config_file_name is not None: - fileConfig(config.config_file_name) -target_metadata = metadata - - -def run_migrations_offline() -> None: - """Run migrations in 'offline' mode.""" - url = config.get_main_option("sqlalchemy.url") - context.configure(url=url, target_metadata=target_metadata, literal_binds=True) - with context.begin_transaction(): - context.run_migrations() - - -def do_run_migrations(connection): - context.configure(connection=connection, target_metadata=target_metadata) - with context.begin_transaction(): - context.run_migrations() - - -async def run_async_migrations(): - connectable = async_engine_from_config( - config.get_section(config.config_ini_section, {}), - prefix="sqlalchemy.", - poolclass=pool.NullPool, - ) - async with connectable.connect() as connection: - await connection.run_sync(do_run_migrations) - await connectable.dispose() - - -def run_migrations_online() -> None: - """Run migrations in 'online' mode.""" - import asyncio - asyncio.run(run_async_migrations()) - - -if context.is_offline_mode(): - run_migrations_offline() -else: - run_migrations_online() diff --git a/apps/osint-dashboard/alembic/script.py.mako b/apps/osint-dashboard/alembic/script.py.mako deleted file mode 100644 index 3776ebd..0000000 --- a/apps/osint-dashboard/alembic/script.py.mako +++ /dev/null @@ -1,28 +0,0 @@ -<%%doc>Template for rendering a Multiple Migration Revision Identifier. -<%%- - from alembic import context - context.configure() --%> -"""${message} - -Revision ID: ${up_revision} -Revises: ${down_revision | comma_n, trim} -Create Date: ${create_date} -""" -from alembic import op -import sqlalchemy as sa -${imports if imports else ""} - -# revision identifiers, used by Alembic. -revision = ${repr(up_revision)} -down_revision = ${repr(down_revision)} -branch_labels = ${repr(branch_labels)} -depends_on = ${repr(depends_on)} - - -def upgrade() -> None: - ${upgrades if upgrades else "pass"} - - -def downgrade() -> None: - ${downgrades if downgrades else "pass"} diff --git a/apps/osint-dashboard/alembic/versions/001_initial.py b/apps/osint-dashboard/alembic/versions/001_initial.py deleted file mode 100644 index 7cc2207..0000000 --- a/apps/osint-dashboard/alembic/versions/001_initial.py +++ /dev/null @@ -1,149 +0,0 @@ -"""initial schema - -Revision ID: 001_initial -Revises: -Create Date: 2026-05-18 -""" - -from alembic import op -import sqlalchemy as sa -from sqlalchemy.dialects.postgresql import UUID, TSVECTOR, ENUM - -# revision identifiers, used by Alembic. -revision = '001_initial' -down_revision = None -branch_labels = None -depends_on = None - - -def upgrade() -> None: - # Enums - op.execute("CREATE TYPE feed_source_type AS ENUM ('rss', 'gdel-t2', 'social', 'earthquake', 'disaster', 'weather', 'fire', 'satellite')") - op.execute("CREATE TYPE event_source_type AS ENUM ('rss', 'gdel-t2', 'social', 'earthquake', 'disaster', 'weather', 'fire', 'satellite')") - op.execute("CREATE TYPE sentiment_label AS ENUM ('positive', 'neutral', 'negative')") - op.execute("CREATE TYPE entity_type AS ENUM ('person', 'organization', 'location', 'topic', 'asset')") - op.execute("CREATE TYPE alert_type AS ENUM ('entity_mention', 'sentiment_shift', 'geo_proximity', 'keyword_match', 'threshold', 'anomaly')") - op.execute("CREATE TYPE alert_severity AS ENUM ('low', 'medium', 'high', 'critical')") - - # Extensions - op.execute("CREATE EXTENSION IF NOT EXISTS postgis") - op.execute("CREATE EXTENSION IF NOT EXISTS timescaledb") - - # feed_sources - op.create_table( - 'feed_sources', - sa.Column('id', UUID(as_uuid=True), primary_key=True), - sa.Column('name', sa.String(256), nullable=False), - sa.Column('source_type', sa.Enum('rss', 'gdel-t2', 'social', 'earthquake', 'disaster', 'weather', 'fire', 'satellite', name='feed_source_type'), nullable=False), - sa.Column('url', sa.Text()), - sa.Column('config', sa.JSON()), - sa.Column('enabled', sa.Integer, server_default='1', nullable=False), - sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), - sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.func.now()), - ) - - # events (will become hypertable) - op.create_table( - 'events', - sa.Column('id', UUID(as_uuid=True), primary_key=True), - sa.Column('source_type', sa.Enum('rss', 'gdel-t2', 'social', 'earthquake', 'disaster', 'weather', 'fire', 'satellite', name='event_source_type'), nullable=False, index=True), - sa.Column('source_id', UUID(as_uuid=True)), - sa.Column('title', sa.Text()), - sa.Column('body', sa.Text()), - sa.Column('url', sa.Text()), - sa.Column('sentiment_score', sa.Float()), - sa.Column('sentiment_label', sa.Enum('positive', 'neutral', 'negative', name='sentiment_label')), - sa.Column('location_lat', sa.Float()), - sa.Column('location_lon', sa.Float()), - sa.Column('location_name', sa.String(512)), - sa.Column('entities', sa.JSON()), - sa.Column('tags', sa.JSON()), - sa.Column('raw', sa.JSON()), - sa.Column('ingested_at', sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), - sa.Column('source_timestamp', sa.DateTime(timezone=True), nullable=False), - sa.Column('search_vector', TSVECTOR), - ) - - # Convert events to TimescaleDB hypertable - op.execute("SELECT create_hypertable('events', 'ingested_at', if_not_exists => TRUE)") - - # GIN index for full-text search - op.create_index('ix_events_search_vector', 'events', ['search_vector'], postgresql_using='gin') - # Spatial index - op.create_index('ix_events_location', 'events', ['location_lat', 'location_lon']) - - # entities - op.create_table( - 'entities', - sa.Column('id', UUID(as_uuid=True), primary_key=True), - sa.Column('name', sa.String(512), nullable=False, index=True), - sa.Column('entity_type', sa.Enum('person', 'organization', 'location', 'topic', 'asset', name='entity_type'), nullable=False), - sa.Column('aliases', sa.JSON()), - sa.Column('description', sa.Text()), - sa.Column('metadata', sa.JSON()), - sa.Column('location_lat', sa.Float()), - sa.Column('location_lon', sa.Float()), - sa.Column('event_count', sa.Integer, server_default='0'), - sa.Column('first_seen', sa.DateTime(timezone=True), server_default=sa.func.now()), - sa.Column('last_seen', sa.DateTime(timezone=True), server_default=sa.func.now()), - ) - - # entity_events - op.create_table( - 'entity_events', - sa.Column('entity_id', UUID(as_uuid=True), primary_key=True), - sa.Column('event_id', UUID(as_uuid=True), primary_key=True), - sa.Column('relevance_score', sa.Float()), - sa.Column('linked_at', sa.DateTime(timezone=True), server_default=sa.func.now()), - ) - - # alerts - op.create_table( - 'alerts', - sa.Column('id', UUID(as_uuid=True), primary_key=True), - sa.Column('alert_type', sa.Enum('entity_mention', 'sentiment_shift', 'geo_proximity', 'keyword_match', 'threshold', 'anomaly', name='alert_type'), nullable=False), - sa.Column('entity_id', UUID(as_uuid=True)), - sa.Column('event_id', UUID(as_uuid=True)), - sa.Column('severity', sa.Enum('low', 'medium', 'high', 'critical', name='alert_severity'), nullable=False), - sa.Column('title', sa.Text(), nullable=False), - sa.Column('message', sa.Text()), - sa.Column('context', sa.JSON()), - sa.Column('acknowledged', sa.Integer, server_default='0'), - sa.Column('acknowledged_by', sa.String(256)), - sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), - sa.Column('resolved_at', sa.DateTime(timezone=True)), - ) - op.create_index('ix_alerts_severity_created', 'alerts', ['severity', 'created_at']) - op.create_index('ix_alerts_entity', 'alerts', ['entity_id']) - - # documents - op.create_table( - 'documents', - sa.Column('id', UUID(as_uuid=True), primary_key=True), - sa.Column('bucket', sa.String(256), nullable=False), - sa.Column('object_key', sa.String(1024), nullable=False), - sa.Column('content_type', sa.String(256)), - sa.Column('size_bytes', sa.Integer()), - sa.Column('description', sa.Text()), - sa.Column('tags', sa.JSON()), - sa.Column('event_id', UUID(as_uuid=True)), - sa.Column('uploaded_at', sa.DateTime(timezone=True), server_default=sa.func.now()), - ) - - -def downgrade() -> None: - op.drop_table('documents') - op.drop_table('alerts') - op.drop_table('entity_events') - op.drop_table('entities') - op.execute("SELECT drop_hypertable('events', cascade => TRUE)") - op.drop_table('events') - op.drop_table('feed_sources') - - # Drop enums - op.execute("DROP TYPE IF EXISTS feed_source_type") - op.execute("DROP TYPE IF EXISTS event_source_type") - op.execute("DROP TYPE IF EXISTS sentiment_label") - op.execute("DROP TYPE IF EXISTS entity_type") - op.execute("DROP TYPE IF EXISTS alert_type") - op.execute("DROP TYPE IF EXISTS alert_severity") diff --git a/apps/osint-dashboard/app/database.py b/apps/osint-dashboard/app/database.py deleted file mode 100644 index 767c26a..0000000 --- a/apps/osint-dashboard/app/database.py +++ /dev/null @@ -1,28 +0,0 @@ -import os - -from sqlalchemy import MetaData, event, text -from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine - -DB_USER = os.getenv("DB_USER", "osint") -DB_PASS = os.getenv("DB_PASSWORD", "") -DB_HOST = os.getenv("DB_HOST", "osint-pgdb-rw.customer1.svc.cluster.local") -DB_PORT = os.getenv("DB_PORT", "5432") -DB_NAME = os.getenv("DB_NAME", "osint_data") - -DATABASE_URL = f"postgresql+asyncpg://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}" - -engine = create_async_engine( - DATABASE_URL, echo=False, pool_size=5, max_overflow=10, pool_recycle=300 -) -async_session = async_sessionmaker( - engine, class_=AsyncSession, expire_on_commit=False -) -metadata = MetaData() - - -async def init_extensions(): - """Initialize PostGIS and TimescaleDB extensions on first connection.""" - async with engine.connect() as conn: - await conn.execute(text("CREATE EXTENSION IF NOT EXISTS postgis")) - await conn.execute(text("CREATE EXTENSION IF NOT EXISTS timescaledb")) - await conn.commit() diff --git a/apps/osint-dashboard/app/ingest_cron.py b/apps/osint-dashboard/app/ingest_cron.py deleted file mode 100644 index a0dcfa6..0000000 --- a/apps/osint-dashboard/app/ingest_cron.py +++ /dev/null @@ -1,46 +0,0 @@ -"""CronJob entry point for scheduled ingestion.""" - -import asyncio -import os -import sys -from pathlib import Path - -# Add app dir to path -sys.path.insert(0, str(Path(__file__).parent)) - -from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes -from ingestor import fetch_and_process - -INGESTOR_TYPE = os.getenv("INGESTOR_TYPE", "rss") -RSS_URL = os.getenv("RSS_URL", "") - - -async def main(): - print(f"Starting ingester: {INGESTOR_TYPE}") - - if INGESTOR_TYPE == "rss": - if not RSS_URL: - print("No RSS_URL set, skipping") - return - count = await ingest_rss_feed(RSS_URL) - print(f"RSS: ingested {count} items") - - elif INGESTOR_TYPE == "gdelt": - count = await ingest_gdelt(max_articles=50) - print(f"GDELT: ingested {count} articles") - - elif INGESTOR_TYPE == "earthquake": - count = await ingest_earthquakes() - print(f"Earthquakes: ingested {count} events") - - elif INGESTOR_TYPE == "nats": - count = await fetch_and_process(batch_size=500) - print(f"NATS: processed {count} messages") - - else: - print(f"Unknown ingestor type: {INGESTOR_TYPE}") - sys.exit(1) - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/apps/osint-dashboard/app/ingestor.py b/apps/osint-dashboard/app/ingestor.py deleted file mode 100644 index d74d4f0..0000000 --- a/apps/osint-dashboard/app/ingestor.py +++ /dev/null @@ -1,107 +0,0 @@ -"""NATS JetStream consumer — ingests OSINT events from NATS streams.""" - -from __future__ import annotations - -import json -import logging -from datetime import datetime, timezone - -import nats -from nats.errors import TimeoutError - -from database import async_session -from models import events as events_table - -logger = logging.getLogger("osint.ingestor") - -# NATS connection settings -NATS_URLS = "nats://osint-nats.customer1.svc.cluster.local:4222" -NATS_STREAM = "events" -NATS_DURABLE = "osint-ingestor" - -# Redis cache settings -REDIS_URL = "redis://osint-redis-sentinel.customer1.svc.cluster.local:26379/0" - - -async def ingest_event(msg: dict): - """Ingest a single event from NATS into PostgreSQL.""" - event_row = { - "source_type": msg.get("source_type", "rss"), - "source_id": msg.get("source_id"), - "title": msg.get("title"), - "body": msg.get("body"), - "url": msg.get("url"), - "sentiment_score": msg.get("sentiment_score"), - "sentiment_label": msg.get("sentiment_label"), - "location_lat": msg.get("location_lat"), - "location_lon": msg.get("location_lon"), - "location_name": msg.get("location_name"), - "entities": msg.get("entities", []), - "tags": msg.get("tags", []), - "raw": msg.get("raw"), - "source_timestamp": msg.get("source_timestamp", datetime.now(timezone.utc).isoformat()), - } - - # Parse timestamp if string - if isinstance(event_row["source_timestamp"], str): - event_row["source_timestamp"] = datetime.fromisoformat(event_row["source_timestamp"]) - - async with async_session() as session: - result = await session.execute(events_table.insert().values(**event_row)) - await session.commit() - event_id = result.inserted_primary_key[0] # type: ignore[union-attr] - logger.info("Ingested event %s from source %s", event_id, msg.get("source_type")) - return event_id - - -async def start_nats_consumer(): - """Start NATS JetStream consumer for OSINT events.""" - nc = await nats.connect(NATS_URLS) - js = nc.jetstream() - - # Create stream if not exists - try: - await js.add_stream( - name=NATS_STREAM, - subjects=[ - "events.gdelt", "events.rss", "events.social", - "events.earthquake", "events.disaster", "events.weather", - "events.fire", "events.satellite", "events.new", "events.alert", - ], - retention=nats.js.api.RetentionPolicy.INTERESTS, - max_msgs=1_000_000, - ) - logger.info("Created NATS stream %s", NATS_STREAM) - except Exception: - logger.debug("Stream %s already exists", NATS_STREAM) - - # Create durable consumer - sub = await js.pull_subscribe( - subject="events.>", - durable_name=NATS_DURABLE, - ) - - logger.info("NATS consumer started, durable=%s", NATS_DURABLE) - return nc, sub - - -async def fetch_and_process(batch_size: int = 100): - """Fetch a batch of messages and process them.""" - nc, sub = await start_nats_consumer() - js = nc.jetstream() - - msgs = await sub.fetch(batch_size, timeout=5) - processed = 0 - - for msg in msgs: - try: - data = json.loads(msg.data) - await ingest_event(data) - await msg.ack() - processed += 1 - except Exception: - logger.error("Failed to process message: %s", msg.data, exc_info=True) - - await nc.close() - logger.info("Processed %d messages in batch", processed) - return processed diff --git a/apps/osint-dashboard/app/main.py b/apps/osint-dashboard/app/main.py deleted file mode 100644 index 01be05b..0000000 --- a/apps/osint-dashboard/app/main.py +++ /dev/null @@ -1,603 +0,0 @@ -"""OSINT Dashboard — FastAPI backend. - -Real-time geospatial OSINT dashboard API: -- Event ingestion via NATS JetStream consumers -- Full-text search across events (PostgreSQL tsvector) -- Entity tracking and relationship mapping -- Alert management -- Document storage (MinIO-backed) -- Sentiment aggregation and timeline analytics -""" - -from __future__ import annotations - -import json -import logging -from datetime import datetime, timedelta, timezone -from decimal import Decimal -from pathlib import Path -from uuid import UUID - -import structlog -from fastapi import FastAPI, HTTPException, Query -from fastapi.responses import FileResponse, HTMLResponse -from sqlalchemy import and_, func, select, text -from sqlalchemy.ext.asyncio import AsyncSession - -from database import async_session, init_extensions -from models import ( - alerts, documents, entities, entity_events, events, feed_sources -) -from schemas import ( - AlertCreate, AlertOut, AlertSeverity, AlertType, AlertUpdate, - DashboardSummary, EntityCreate, EntityKind, EntityOut, - EventCreate, EventOut, - FeedSourceCreate, FeedSourceOut, - SearchResult, SentimentSummary, SourceType, - SearchQuery, TimelinePoint, -) -from ingestor import ingest_event, fetch_and_process -from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes, ingest_social_signals - -logging.basicConfig(level=logging.INFO) -logger = structlog.get_logger("osint.dashboard") - -app = FastAPI( - title="OSINT Dashboard", - description="Real-time geospatial OSINT intelligence dashboard", - version="0.1.0", -) - -STATIC_DIR = Path(__file__).parent / "static" - - -# ── Helpers ─────────────────────────────────────────────────────────────── - -def event_to_out(row: dict) -> EventOut: - """Convert DB row dict to EventOut schema.""" - return EventOut( - id=row["id"], - source_type=row["source_type"], - source_id=row["source_id"], - title=row["title"], - body=row["body"], - url=row["url"], - sentiment_score=row["sentiment_score"], - sentiment_label=row["sentiment_label"], - location_lat=row["location_lat"], - location_lon=row["location_lon"], - location_name=row["location_name"], - entities=row["entities"], - tags=row["tags"], - ingested_at=row["ingested_at"], - source_timestamp=row["source_timestamp"], - ) - - -def entity_to_out(row: dict) -> EntityOut: - """Convert DB row dict to EntityOut schema.""" - return EntityOut( - id=row["id"], - name=row["name"], - entity_type=row["entity_type"], - aliases=row["aliases"], - description=row["description"], - metadata=row["metadata"], - location_lat=row["location_lat"], - location_lon=row["location_lon"], - event_count=row["event_count"], - first_seen=row["first_seen"], - last_seen=row["last_seen"], - ) - - -def alert_to_out(row: dict) -> AlertOut: - """Convert DB row dict to AlertOut schema.""" - return AlertOut( - id=row["id"], - alert_type=row["alert_type"], - entity_id=row["entity_id"], - event_id=row["event_id"], - severity=row["severity"], - title=row["title"], - message=row["message"], - context=row["context"], - acknowledged=bool(row["acknowledged"]), - acknowledged_by=row["acknowledged_by"], - created_at=row["created_at"], - resolved_at=row["resolved_at"], - ) - - -# ── Health ──────────────────────────────────────────────────────────────── - -@app.get("/api/health") -async def health(): - """Health check with database connectivity.""" - async with async_session() as session: - result = await session.execute(select(func.now())) - db_time = result.scalar() - return {"status": "ok", "db_time": db_time.isoformat() if db_time else None} - - -# ── Startup ─────────────────────────────────────────────────────────────── - -@app.on_event("startup") -async def startup(): - """Initialize extensions and run migrations.""" - await init_extensions() - from alembic import command - from alembic.config import Config - alembic_cfg = Config(str(Path(__file__).parent.parent / "alembic.ini")) - command.upgrade(alembic_cfg, "head") - - -# ── Feed Sources ────────────────────────────────────────────────────────── - -@app.get("/api/sources", response_model=list[FeedSourceOut]) -async def list_sources(enabled_only: bool = Query(True)): - """List all configured feed sources.""" - async with async_session() as session: - stmt = select(feed_sources).order_by(feed_sources.c.name) - if enabled_only: - stmt = stmt.where(feed_sources.c.enabled == 1) - rows = (await session.execute(stmt)).mappings().all() - return [FeedSourceOut( - id=r["id"], name=r["name"], source_type=r["source_type"], - url=r["url"], config=r["config"], enabled=bool(r["enabled"]), - created_at=r["created_at"], - ) for r in rows] - - -@app.post("/api/sources", status_code=201) -async def create_source(payload: FeedSourceCreate): - """Add a new feed source.""" - async with async_session() as session: - values = payload.model_dump() - result = await session.execute(feed_sources.insert().values(**values)) - await session.commit() - pk = result.inserted_primary_key[0] # type: ignore - return {"id": str(pk)} - - -@app.patch("/api/sources/{source_id}") -async def update_source(source_id: UUID, payload: dict): - """Update a feed source (e.g., toggle enabled).""" - async with async_session() as session: - row = (await session.execute( - select(feed_sources).where(feed_sources.c.id == source_id) - )).mappings().one_or_none() - if not row: - raise HTTPException(404, "Source not found") - await session.execute( - feed_sources.update() - .where(feed_sources.c.id == source_id) - .values(**payload) - ) - await session.commit() - return {"ok": True} - - -# ── Events ──────────────────────────────────────────────────────────────── - -@app.get("/api/events", response_model=list[EventOut]) -async def list_events( - source_type: SourceType | None = Query(None), - limit: int = Query(50, ge=1, le=500), - offset: int = Query(0, ge=0), -): - """List recent ingested events.""" - async with async_session() as session: - stmt = select(events).order_by(events.c.ingested_at.desc()) - if source_type: - stmt = stmt.where(events.c.source_type == source_type.value) - stmt = stmt.limit(limit).offset(offset) - rows = (await session.execute(stmt)).mappings().all() - return [event_to_out(r) for r in rows] - - -@app.get("/api/events/{event_id}", response_model=EventOut) -async def get_event(event_id: UUID): - """Get a single event by ID.""" - async with async_session() as session: - row = (await session.execute( - select(events).where(events.c.id == event_id) - )).mappings().one_or_none() - if not row: - raise HTTPException(404, "Event not found") - return event_to_out(row) - - -@app.post("/api/events", status_code=201) -async def create_event(payload: EventCreate): - """Manually ingest an event (bypasses NATS).""" - values = payload.model_dump(exclude_unset=True) - if not values.get("source_timestamp"): - values["source_timestamp"] = datetime.now(timezone.utc) - event_id = await ingest_event(values) - return {"id": str(event_id)} - - -# ── Search ──────────────────────────────────────────────────────────────── - -@app.post("/api/search", response_model=SearchResult) -async def search_events(query: SearchQuery): - """Full-text search across events with optional filters.""" - async with async_session() as session: - # Build query with tsvector full-text search (parameterized to avoid SQL injection) - tsquery_param = text("plainto_tsquery('english', :q)") - - base_stmt = select( - events, - func.count().over().label("total") - ).where( - events.c.search_vector.op("@@")(tsquery_param) - ) - - # Apply filters - if query.source_type: - base_stmt = base_stmt.where(events.c.source_type == query.source_type.value) - if query.entity_id: - base_stmt = base_stmt.join( - entity_events, entity_events.c.event_id == events.c.id - ).where(entity_events.c.entity_id == query.entity_id) - if query.sentiment: - base_stmt = base_stmt.where(events.c.sentiment_label == query.sentiment.value) - if query.min_date: - base_stmt = base_stmt.where(events.c.source_timestamp >= query.min_date) - if query.max_date: - base_stmt = base_stmt.where(events.c.source_timestamp <= query.max_date) - if query.min_lat is not None and query.max_lat is not None: - base_stmt = base_stmt.where( - and_( - events.c.location_lat >= query.min_lat, - events.c.location_lat <= query.max_lat, - ) - ) - if query.min_lon is not None and query.max_lon is not None: - base_stmt = base_stmt.where( - and_( - events.c.location_lon >= query.min_lon, - events.c.location_lon <= query.max_lon, - ) - ) - - base_stmt = base_stmt.order_by(events.c.ingested_at.desc()) - base_stmt = base_stmt.limit(query.limit).offset(query.offset) - - result = (await session.execute(base_stmt, {"q": query.q})).mappings().all() - if result: - total = result[0]["total"] - else: - total = 0 - evts = [event_to_out(r) for r in result] - - return SearchResult( - events=evts, - total=total, - has_more=query.offset + len(evts) < total, - ) - - -# ── Entities ────────────────────────────────────────────────────────────── - -@app.get("/api/entities", response_model=list[EntityOut]) -async def list_entities( - entity_type: EntityKind | None = Query(None), - limit: int = Query(50, ge=1, le=500), -): - """List tracked entities.""" - async with async_session() as session: - stmt = select(entities).order_by(entities.c.event_count.desc()) - if entity_type: - stmt = stmt.where(entities.c.entity_type == entity_type.value) - stmt = stmt.limit(limit) - rows = (await session.execute(stmt)).mappings().all() - return [entity_to_out(r) for r in rows] - - -@app.get("/api/entities/{entity_id}", response_model=EntityOut) -async def get_entity(entity_id: UUID): - """Get entity details with recent events.""" - async with async_session() as session: - row = (await session.execute( - select(entities).where(entities.c.id == entity_id) - )).mappings().one_or_none() - if not row: - raise HTTPException(404, "Entity not found") - return entity_to_out(row) - - -@app.post("/api/entities", status_code=201) -async def create_entity(payload: EntityCreate): - """Create or update a tracked entity.""" - async with async_session() as session: - # Check if entity already exists by name - existing = (await session.execute( - select(entities).where(entities.c.name == payload.name) - )).mappings().one_or_none() - - if existing: - # Update - updates = payload.model_dump(exclude_unset=True) - updates["last_seen"] = datetime.now(timezone.utc) - await session.execute( - entities.update() - .where(entities.c.id == existing["id"]) - .values(**updates) - ) - await session.commit() - return {"id": str(existing["id"]), "created": False} - - # Create - values = payload.model_dump() - result = await session.execute(entities.insert().values(**values)) - await session.commit() - pk = result.inserted_primary_key[0] # type: ignore - return {"id": str(pk), "created": True} - - -@app.get("/api/entities/{entity_id}/events", response_model=list[EventOut]) -async def get_entity_events( - entity_id: UUID, - limit: int = Query(50, ge=1, le=500), -): - """Get events linked to a specific entity.""" - async with async_session() as session: - stmt = ( - select(events) - .join(entity_events, entity_events.c.event_id == events.c.id) - .where(entity_events.c.entity_id == entity_id) - .order_by(events.c.source_timestamp.desc()) - .limit(limit) - ) - rows = (await session.execute(stmt)).mappings().all() - return [event_to_out(r) for r in rows] - - -# ── Alerts ──────────────────────────────────────────────────────────────── - -@app.get("/api/alerts", response_model=list[AlertOut]) -async def list_alerts( - severity: AlertSeverity | None = Query(None), - acknowledged: bool | None = Query(None), - entity_id: UUID | None = Query(None), - limit: int = Query(50, ge=1, le=500), -): - """List alerts with optional filters.""" - async with async_session() as session: - stmt = select(alerts).order_by( - alerts.c.severity.desc(), alerts.c.created_at.desc() - ) - if severity: - stmt = stmt.where(alerts.c.severity == severity.value) - if acknowledged is not None: - stmt = stmt.where(alerts.c.acknowledged == int(acknowledged)) - if entity_id: - stmt = stmt.where(alerts.c.entity_id == entity_id) - stmt = stmt.limit(limit) - rows = (await session.execute(stmt)).mappings().all() - return [alert_to_out(r) for r in rows] - - -@app.post("/api/alerts", status_code=201) -async def create_alert(payload: AlertCreate): - """Create a new alert.""" - async with async_session() as session: - values = payload.model_dump() - result = await session.execute(alerts.insert().values(**values)) - await session.commit() - pk = result.inserted_primary_key[0] # type: ignore - return {"id": str(pk)} - - -@app.patch("/api/alerts/{alert_id}") -async def update_alert(alert_id: UUID, payload: AlertUpdate): - """Update alert (acknowledge, resolve).""" - async with async_session() as session: - row = (await session.execute( - select(alerts).where(alerts.c.id == alert_id) - )).mappings().one_or_none() - if not row: - raise HTTPException(404, "Alert not found") - updates = payload.model_dump(exclude_unset=True) - if "acknowledged" in updates: - updates["acknowledged"] = int(updates["acknowledged"]) - await session.execute( - alerts.update().where(alerts.c.id == alert_id).values(**updates) - ) - await session.commit() - return {"ok": True} - - -# ── Documents ───────────────────────────────────────────────────────────── - -@app.get("/api/documents", response_model=dict) -async def list_documents( - limit: int = Query(50, ge=1, le=500), - offset: int = Query(0, ge=0), -): - """List documents indexed in MinIO.""" - async with async_session() as session: - stmt = select(documents).order_by(documents.c.uploaded_at.desc()).limit(limit).offset(offset) - rows = (await session.execute(stmt)).mappings().all() - return { - "documents": [{ - "id": str(r["id"]), "bucket": r["bucket"], "object_key": r["object_key"], - "content_type": r["content_type"], "size_bytes": r["size_bytes"], - "description": r["description"], "tags": r["tags"], - "event_id": str(r["event_id"]) if r["event_id"] else None, - "uploaded_at": r["uploaded_at"].isoformat() if r["uploaded_at"] else None, - } for r in rows], - } - - -# ── Ingestion Triggers ─────────────────────────────────────────────────── - -@app.post("/api/ingest/rss") -async def trigger_rss_ingest(feed_url: str, source_id: str | None = None): - """Trigger RSS feed ingestion.""" - count = await ingest_rss_feed(feed_url, source_id) - return {"status": "ok", "items_ingested": count} - - -@app.post("/api/ingest/gdelt") -async def trigger_gdelt_ingest(query: str = "", max_articles: int = 50): - """Trigger GDELT data ingestion.""" - count = await ingest_gdelt(query, max_articles) - return {"status": "ok", "articles_ingested": count} - - -@app.post("/api/ingest/earthquakes") -async def trigger_earthquake_ingest(): - """Trigger USGS earthquake ingestion.""" - count = await ingest_earthquakes() - return {"status": "ok", "events_ingested": count} - - -@app.post("/api/ingest/social") -async def trigger_social_ingest(query: str = "", max_items: int = 50): - """Trigger social signals ingestion.""" - count = await ingest_social_signals(query, max_items) - return {"status": "ok", "signals_ingested": count} - - -@app.post("/api/ingest/process") -async def trigger_nats_processing(batch_size: int = 100): - """Process pending NATS JetStream messages.""" - count = await fetch_and_process(batch_size) - return {"status": "ok", "processed": count} - - -# ── Analytics / Aggregation ─────────────────────────────────────────────── - -@app.get("/api/analytics/summary", response_model=DashboardSummary) -async def get_dashboard_summary(): - """Dashboard overview: event counts, sentiment, top entities, alerts.""" - async with async_session() as session: - now = datetime.now(timezone.utc) - yesterday = now - timedelta(hours=24) - - # Total events - total = (await session.execute( - select(func.count()).select_from(events) - )).scalar() or 0 - - # Events in last 24h - events_24h = (await session.execute( - select(func.count()).where(events.c.ingested_at >= yesterday) - )).scalar() or 0 - - # Active sources - active = (await session.execute( - select(func.count()).where(feed_sources.c.enabled == 1) - )).scalar() or 0 - - # Open alerts - open_alerts = (await session.execute( - select(func.count()).where(alerts.c.acknowledged == 0) - )).scalar() or 0 - - # Tracked entities - ent_count = (await session.execute( - select(func.count()).select_from(entities) - )).scalar() or 0 - - # Sentiment breakdown (last 24h) - def sentiment_query(): - return select( - func.count().where(events.c.sentiment_label == "positive").label("pos"), - func.count().where(events.c.sentiment_label == "neutral").label("neu"), - func.count().where(events.c.sentiment_label == "negative").label("neg"), - func.avg(events.c.sentiment_score).label("avg"), - ).where(events.c.ingested_at >= yesterday) - - sent_row = (await session.execute(sentiment_query())).mappings().one() - sentiment = SentimentSummary( - period="24h", - positive_count=sent_row["pos"] or 0, - neutral_count=sent_row["neu"] or 0, - negative_count=sent_row["neg"] or 0, - avg_score=float(sent_row["avg"] or 0), - ) - - # Top entities by event count - top_ent = (await session.execute( - select(entities).order_by(entities.c.event_count.desc()).limit(10) - )).mappings().all() - - return DashboardSummary( - total_events=total, - events_last_24h=events_24h, - active_sources=active, - open_alerts=open_alerts, - tracked_entities=ent_count, - sentiment=sentiment, - top_entities=[entity_to_out(r) for r in top_ent], - ) - - -@app.get("/api/analytics/timeline") -async def get_timeline( - hours: int = Query(24, ge=1, le=168), - bucket_hours: int = Query(1, ge=1, le=24), -): - """Event timeline: counts and avg sentiment per time bucket.""" - async with async_session() as session: - cutoff = datetime.now(timezone.utc) - timedelta(hours=hours) - # Use date_trunc for bucketing - buckets = await session.execute(text(f""" - SELECT - date_trunc('hour', source_timestamp) AS ts, - COUNT(*) AS event_count, - COALESCE(AVG(sentiment_score), 0) AS avg_sentiment - FROM events - WHERE source_timestamp >= :cutoff - GROUP BY ts - ORDER BY ts - """), {"cutoff": cutoff}) - rows = buckets.mappings().all() - - return [TimelinePoint(timestamp=r["ts"], event_count=r["event_count"], - avg_sentiment=float(r["avg_sentiment"])) for r in rows] - - -@app.get("/api/analytics/sentiment/by-source") -async def sentiment_by_source(hours: int = 24): - """Sentiment breakdown grouped by source type.""" - async with async_session() as session: - cutoff = datetime.now(timezone.utc) - timedelta(hours=hours) - result = await session.execute(text(f""" - SELECT - source_type, - COUNT(*) AS total, - COUNT(*) FILTER (WHERE sentiment_label = 'positive') AS positive, - COUNT(*) FILTER (WHERE sentiment_label = 'neutral') AS neutral, - COUNT(*) FILTER (WHERE sentiment_label = 'negative') AS negative, - COALESCE(AVG(sentiment_score), 0) AS avg_score - FROM events - WHERE ingested_at >= :cutoff - GROUP BY source_type - ORDER BY total DESC - """), {"cutoff": cutoff}) - rows = result.mappings().all() - return [{ - "source_type": r["source_type"], - "total": r["total"], - "positive": r["positive"], - "neutral": r["neutral"], - "negative": r["negative"], - "avg_score": float(r["avg_score"]), - } for r in rows] - - -# ── Frontend ────────────────────────────────────────────────────────────── - -@app.get("/", response_class=HTMLResponse) -async def index(): - return FileResponse(str(STATIC_DIR / "index.html")) - - -if __name__ == "__main__": - import uvicorn - uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/apps/osint-dashboard/app/models.py b/apps/osint-dashboard/app/models.py deleted file mode 100644 index 6f052b2..0000000 --- a/apps/osint-dashboard/app/models.py +++ /dev/null @@ -1,147 +0,0 @@ -"""OSINT Dashboard — SQLAlchemy models (async, declarative).""" - -from datetime import datetime, timezone -from uuid import uuid4 - -from sqlalchemy import ( - Column, Enum, Float, Index, Integer, String, Text, - DateTime, JSON, func, Table, -) -from sqlalchemy.dialects.postgresql import UUID, TSVECTOR - -from database import metadata - - -# ── Feed Sources ────────────────────────────────────────────────────────── - -feed_sources = Table( - "feed_sources", - metadata, - Column("id", UUID(as_uuid=True), primary_key=True, default=uuid4), - Column("name", String(256), nullable=False), - Column("source_type", Enum( - "rss", "gdel-t2", "social", "earthquake", "disaster", - "weather", "fire", "satellite", name="feed_source_type" - ), nullable=False), - Column("url", Text), - Column("config", JSON), - Column("enabled", Integer, server_default="1", nullable=False), - Column("created_at", DateTime(timezone=True), server_default=func.now(), nullable=False), - Column("updated_at", DateTime(timezone=True), server_default=func.now(), onupdate=func.now()), -) - - -# ── Events (hypertable via TimescaleDB) ────────────────────────────────── - -events = Table( - "events", - metadata, - Column("id", UUID(as_uuid=True), primary_key=True, default=uuid4), - Column("source_type", Enum( - "rss", "gdel-t2", "social", "earthquake", "disaster", - "weather", "fire", "satellite", name="event_source_type" - ), nullable=False, index=True), - Column("source_id", UUID(as_uuid=True)), - Column("title", Text), - Column("body", Text), - Column("url", Text), - Column("sentiment_score", Float), - Column("sentiment_label", Enum("positive", "neutral", "negative", name="sentiment_label")), - Column("location_lat", Float), - Column("location_lon", Float), - Column("location_name", String(512)), - Column("entities", JSON), - Column("tags", JSON), - Column("raw", JSON), - Column("ingested_at", DateTime(timezone=True), server_default=func.now(), nullable=False), - Column("source_timestamp", DateTime(timezone=True), nullable=False), - # Full-text search vector - Column( - "search_vector", - TSVECTOR, - nullable=True, - ), -) - -# GIN index for full-text search -Index("ix_events_search_vector", events.c.search_vector, postgresql_using="gin") -# Spatial index on location -Index("ix_events_location", events.c.location_lat, events.c.location_lon) - - -# ── Entities (people, organizations, locations of interest) ────────────── - -entities = Table( - "entities", - metadata, - Column("id", UUID(as_uuid=True), primary_key=True, default=uuid4), - Column("name", String(512), nullable=False, index=True), - Column("entity_type", Enum( - "person", "organization", "location", "topic", "asset", - name="entity_type" - ), nullable=False), - Column("aliases", JSON), - Column("description", Text), - Column("metadata", JSON), - Column("location_lat", Float), - Column("location_lon", Float), - Column("event_count", Integer, server_default="0"), - Column("first_seen", DateTime(timezone=True), server_default=func.now()), - Column("last_seen", DateTime(timezone=True), server_default=func.now()), -) - - -# ── Entity-Event Link ──────────────────────────────────────────────────── - -entity_events = Table( - "entity_events", - metadata, - Column("entity_id", UUID(as_uuid=True), primary_key=True), - Column("event_id", UUID(as_uuid=True), primary_key=True), - Column("relevance_score", Float), - Column("linked_at", DateTime(timezone=True), server_default=func.now()), -) - - -# ── Alerts ─────────────────────────────────────────────────────────────── - -alerts = Table( - "alerts", - metadata, - Column("id", UUID(as_uuid=True), primary_key=True, default=uuid4), - Column("alert_type", Enum( - "entity_mention", "sentiment_shift", "geo_proximity", - "keyword_match", "threshold", "anomaly", - name="alert_type" - ), nullable=False), - Column("entity_id", UUID(as_uuid=True)), - Column("event_id", UUID(as_uuid=True)), - Column("severity", Enum("low", "medium", "high", "critical", name="alert_severity"), nullable=False), - Column("title", Text, nullable=False), - Column("message", Text), - Column("context", JSON), - Column("acknowledged", Integer, server_default="0"), - Column("acknowledged_by", String(256)), - Column("created_at", DateTime(timezone=True), server_default=func.now(), nullable=False), - Column("resolved_at", DateTime(timezone=True)), -) - -Index("ix_alerts_severity_created", alerts.c.severity, alerts.c.created_at.desc()) -Index("ix_alerts_entity", alerts.c.entity_id) - - -# ── Documents (stored in MinIO, indexed here) ──────────────────────────── - -documents = Table( - "documents", - metadata, - Column("id", UUID(as_uuid=True), primary_key=True, default=uuid4), - Column("bucket", String(256), nullable=False), - Column("object_key", String(1024), nullable=False), - Column("content_type", String(256)), - Column("size_bytes", Integer), - Column("description", Text), - Column("tags", JSON), - Column("event_id", UUID(as_uuid=True)), - Column("uploaded_at", DateTime(timezone=True), server_default=func.now()), -) diff --git a/apps/osint-dashboard/app/requirements.txt b/apps/osint-dashboard/app/requirements.txt deleted file mode 100644 index 3edff72..0000000 --- a/apps/osint-dashboard/app/requirements.txt +++ /dev/null @@ -1,13 +0,0 @@ -fastapi>=0.115 -uvicorn[standard]>=0.34 -sqlalchemy[asyncio]>=2.0 -asyncpg>=0.30 -nats-py>=2.9 -redis[hiredis]>=5.2 -minio>=7.2 -pydantic>=2.10 -alembic>=1.14 -httpx>=0.28 -feedparser>=6.0 -python-dateutil>=2.9 -structlog>=24.4 diff --git a/apps/osint-dashboard/app/schemas.py b/apps/osint-dashboard/app/schemas.py deleted file mode 100644 index 1cac012..0000000 --- a/apps/osint-dashboard/app/schemas.py +++ /dev/null @@ -1,242 +0,0 @@ -"""OSINT Dashboard — Pydantic schemas.""" - -from __future__ import annotations - -from datetime import datetime -from enum import Enum -from typing import Optional -from uuid import UUID - -from pydantic import BaseModel, Field - - -# ─── Enums ─────────────────────────────────────────────────────────────── - -class SourceType(str, Enum): - rss = "rss" - gdel_t2 = "gdel-t2" - social = "social" - earthquake = "earthquake" - disaster = "disaster" - weather = "weather" - fire = "fire" - satellite = "satellite" - - -class EntityKind(str, Enum): - person = "person" - organization = "organization" - location = "location" - topic = "topic" - asset = "asset" - - -class Sentiment(str, Enum): - positive = "positive" - neutral = "neutral" - negative = "negative" - - -class AlertType(str, Enum): - entity_mention = "entity_mention" - sentiment_shift = "sentiment_shift" - geo_proximity = "geo_proximity" - keyword_match = "keyword_match" - threshold = "threshold" - anomaly = "anomaly" - - -class AlertSeverity(str, Enum): - low = "low" - medium = "medium" - high = "high" - critical = "critical" - - -# ─── Feed Sources ─────────────────────────────────────────────────────── - -class FeedSourceCreate(BaseModel): - name: str - source_type: SourceType - url: Optional[str] = None - config: Optional[dict] = None - - -class FeedSourceOut(BaseModel): - id: UUID - name: str - source_type: SourceType - url: Optional[str] - config: Optional[dict] - enabled: bool - created_at: datetime - - -# ─── Events ───────────────────────────────────────────────────────────── - -class EventCreate(BaseModel): - source_type: SourceType - source_id: Optional[UUID] = None - title: Optional[str] = None - body: Optional[str] = None - url: Optional[str] = None - sentiment_score: Optional[float] = None - sentiment_label: Optional[Sentiment] = None - location_lat: Optional[float] = None - location_lon: Optional[float] = None - location_name: Optional[str] = None - entities: Optional[list[dict]] = None - tags: Optional[list[str]] = None - raw: Optional[dict] = None - source_timestamp: Optional[datetime] = None - - -class EventOut(BaseModel): - id: UUID - source_type: SourceType - source_id: Optional[UUID] - title: Optional[str] - body: Optional[str] - url: Optional[str] - sentiment_score: Optional[float] - sentiment_label: Optional[Sentiment] - location_lat: Optional[float] - location_lon: Optional[float] - location_name: Optional[str] - entities: Optional[list[dict]] - tags: Optional[list[str]] - ingested_at: datetime - source_timestamp: datetime - - -# ─── Search ────────────────────────────────────────────────────────────── - -class SearchQuery(BaseModel): - q: str = Field(..., min_length=1, max_length=500) - source_type: Optional[SourceType] = None - entity_id: Optional[UUID] = None - sentiment: Optional[Sentiment] = None - min_date: Optional[datetime] = None - max_date: Optional[datetime] = None - min_lat: Optional[float] = None - max_lat: Optional[float] = None - min_lon: Optional[float] = None - max_lon: Optional[float] = None - limit: int = Field(50, ge=1, le=500) - offset: int = Field(0, ge=0) - - -class SearchResult(BaseModel): - events: list[EventOut] - total: int - has_more: bool - - -# ─── Entities ──────────────────────────────────────────────────────────── - -class EntityCreate(BaseModel): - name: str - entity_type: EntityKind - aliases: Optional[list[str]] = None - description: Optional[str] = None - metadata: Optional[dict] = None - location_lat: Optional[float] = None - location_lon: Optional[float] = None - - -class EntityOut(BaseModel): - id: UUID - name: str - entity_type: EntityKind - aliases: Optional[list[str]] - description: Optional[str] - metadata: Optional[dict] - location_lat: Optional[float] - location_lon: Optional[float] - event_count: int - first_seen: datetime - last_seen: datetime - - -# ─── Alerts ────────────────────────────────────────────────────────────── - -class AlertCreate(BaseModel): - alert_type: AlertType - entity_id: Optional[UUID] = None - event_id: Optional[UUID] = None - severity: AlertSeverity - title: str - message: Optional[str] = None - context: Optional[dict] = None - - -class AlertUpdate(BaseModel): - acknowledged: Optional[bool] = None - acknowledged_by: Optional[str] = None - resolved_at: Optional[datetime] = None - - -class AlertOut(BaseModel): - id: UUID - alert_type: AlertType - entity_id: Optional[UUID] - event_id: Optional[UUID] - severity: AlertSeverity - title: str - message: Optional[str] - context: Optional[dict] - acknowledged: bool - acknowledged_by: Optional[str] - created_at: datetime - resolved_at: Optional[datetime] - - -# ─── Documents ─────────────────────────────────────────────────────────── - -class DocumentCreate(BaseModel): - bucket: str - object_key: str - content_type: Optional[str] = None - size_bytes: Optional[int] = None - description: Optional[str] = None - tags: Optional[list[str]] = None - event_id: Optional[UUID] = None - - -class DocumentOut(BaseModel): - id: UUID - bucket: str - object_key: str - content_type: Optional[str] - size_bytes: Optional[int] - description: Optional[str] - tags: Optional[list[str]] - event_id: Optional[UUID] - uploaded_at: datetime - - -# ─── Aggregations ──────────────────────────────────────────────────────── - -class SentimentSummary(BaseModel): - period: str - positive_count: int - neutral_count: int - negative_count: int - avg_score: float - - -class TimelinePoint(BaseModel): - timestamp: datetime - event_count: int - avg_sentiment: float - - -class DashboardSummary(BaseModel): - total_events: int - events_last_24h: int - active_sources: int - open_alerts: int - tracked_entities: int - sentiment: SentimentSummary - top_entities: list[EntityOut] - diff --git a/apps/osint-dashboard/app/sources.py b/apps/osint-dashboard/app/sources.py deleted file mode 100644 index 8ca87ff..0000000 --- a/apps/osint-dashboard/app/sources.py +++ /dev/null @@ -1,186 +0,0 @@ -"""Data source ingestors — fetch from external APIs and push to NATS.""" - -from __future__ import annotations - -import json -import logging -from datetime import datetime, timezone -from email.utils import parsedate_to_datetime - -import httpx -import feedparser -import nats - -logger = logging.getLogger("osint.sources") - - -def _parse_rfc822(date_str: object) -> str | None: - """Parse RFC-822 date string from feedparser entries.""" - if not isinstance(date_str, str): - return None - try: - return parsedate_to_datetime(date_str).astimezone(timezone.utc).isoformat() - except (ValueError, TypeError): - return None - -# NATS connection -NATS_URLS = "nats://osint-nats.customer1.svc.cluster.local:4222" - - -async def publish_event(subject: str, event: dict): - """Publish an event to NATS JetStream.""" - nc = await nats.connect(NATS_URLS) - js = nc.jetstream() - await js.publish(subject, json.dumps(event).encode()) - await nc.close() - logger.debug("Published event to %s", subject) - - -# ─── RSS Feed Ingestor ────────────────────────────────────────────────── - -async def ingest_rss_feed(feed_url: str, source_id: str = None): - """Fetch and parse an RSS feed, publish items to NATS.""" - async with httpx.AsyncClient(timeout=30) as client: - resp = await client.get(feed_url) - resp.raise_for_status() - feed = feedparser.parse(resp.text) - - count = 0 - for entry in feed.entries[:100]: # max 100 per run - event = { - "source_type": "rss", - "source_id": source_id, - "title": entry.get("title"), - "body": entry.get("summary") or entry.get("description"), - "url": entry.get("link"), - "source_timestamp": _parse_rfc822(entry.get("published")) - or datetime.now(timezone.utc).isoformat(), - "tags": [t.get("term") for t in entry.get("tags", []) if t.get("term")], - "raw": { - "feed_title": feed.feed.get("title"), - "author": entry.get("author"), - "categories": [c.get("term") for c in entry.get("categories", [])], - }, - } - await publish_event("events.rss", event) - count += 1 - - logger.info("Ingested %d items from RSS feed %s", count, feed_url) - return count - - -# ─── GDELT 2.0 Ingestor ───────────────────────────────────────────────── - -GDELT_API = "https://api.gdeltproject.org/gdeltv2" - - -async def ingest_gdelt(query: str = "", max_articles: int = 50): - """Fetch articles from GDELT 2.0 API.""" - params = { - "mode": "artlist", - "format": "json", - "maxrecords": max_articles, - "mode": "artlist", - } - if query: - params["search"] = query - - async with httpx.AsyncClient(timeout=60) as client: - resp = await client.get(GDELT_API, params=params) - resp.raise_for_status() - data = resp.json() - - count = 0 - for article in data.get("articles", []): - event = { - "source_type": "gdel-t2", - "title": article.get("title"), - "body": article.get("articleBody"), - "url": article.get("url"), - "sentiment_score": _parse_gdelt_tone(article.get("Tone", "0")), - "location_lat": article.get("Latitude"), - "location_lon": article.get("Longitude"), - "location_name": article.get("Location"), - "source_timestamp": article.get("FirstCreated"), - "entities": [ - {"name": e.get("Topic"), "type": "topic"} - for e in article.get("Mentions", []) - if e.get("Topic") - ], - "raw": article, - } - await publish_event("events.gdelt", event) - count += 1 - - logger.info("Ingested %d articles from GDELT", count) - return count - - -def _parse_gdelt_tone(tone: str) -> float | None: - """Parse GDELT tone string to a -1..1 sentiment score.""" - try: - tone_float = float(tone) - return max(-1.0, min(1.0, tone_float / 4249.0)) # GDELT tone range - except (ValueError, TypeError): - return None - - -# ─── Earthquake Ingestor (USGS) ───────────────────────────────────────── - -USGS_API = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_hour.geojson" - - -async def ingest_earthquakes(): - """Fetch recent earthquakes from USGS.""" - async with httpx.AsyncClient(timeout=30) as client: - resp = await client.get(USGS_API) - resp.raise_for_status() - data = resp.json() - - count = 0 - for feature in data.get("features", []): - props = feature.get("properties", {}) - geometry = feature.get("geometry", {}).get("coordinates", []) - event = { - "source_type": "earthquake", - "title": props.get("title"), - "body": props.get("description"), - "url": props.get("url"), - "location_lat": geometry[1] if len(geometry) > 1 else None, - "location_lon": geometry[0] if len(geometry) > 0 else None, - "location_name": props.get("place"), - "sentiment_label": "neutral", - "tags": [f"magnitude:{props.get('mag')}"] if props.get("mag") else [], - "source_timestamp": ( - datetime.utcfromtimestamp(props.get("time", 0) / 1000) - .replace(tzinfo=timezone.utc) - .isoformat() - ), - "raw": props, - } - await publish_event("events.earthquake", event) - count += 1 - - logger.info("Ingested %d earthquake events", count) - return count - - -# ─── Social Signals (Twitter/X-like placeholder) ──────────────────────── - -async def ingest_social_signals(query: str = "", max_items: int = 50): - """Placeholder for social media signal ingestion. - - In production, this would connect to Twitter API, Reddit, NewsAPI, etc. - For now, it publishes a heartbeat to signal the pipeline is active. - """ - event = { - "source_type": "social", - "title": f"Social signal scan: {query}", - "body": f"Scanned for '{query}' — placeholder connector", - "source_timestamp": datetime.now(timezone.utc).isoformat(), - "tags": [query] if query else [], - "raw": {"query": query, "max_items": max_items, "connector": "placeholder"}, - } - await publish_event("events.social", event) - logger.info("Social signal scan complete for '%s'", query) - return 1 diff --git a/apps/osint-dashboard/app/static/index.html b/apps/osint-dashboard/app/static/index.html deleted file mode 100644 index dbf3f8b..0000000 --- a/apps/osint-dashboard/app/static/index.html +++ /dev/null @@ -1,307 +0,0 @@ - - - - - - OSINT Dashboard - - - -
-

OSINT Dashboard

-
- Status: checking... -  |  Last update: - -
-
- -
- -
-

Total Events

-
-

Events (24h)

-
last 24 hours
-

Active Sources

-
-

Open Alerts

-
-

Tracked Entities

-
-
-

Sentiment (24h)

-
-
-
-
-
-
-
-
- - -
-

Search Events

-
- - -
-
- - -
- - - - -
- - -
-

Recent Events

- - - -
TimeSourceTitleSentimentLocation
-
- - - - - - - - - - - - -
- - - - diff --git a/trading-scripts/README.md b/trading-scripts/README.md deleted file mode 100644 index a7d519d..0000000 --- a/trading-scripts/README.md +++ /dev/null @@ -1,91 +0,0 @@ -# Trading Scripts: Market Data Fetcher - -Historical market data pipeline for AI analysis. Fetches stocks/futures/crypto, computes stats/charts, exports for uncensored LLMs. - -## 🚀 Quick Start - -```bash -# Install deps (one-time) -pip install yfinance pandas matplotlib seaborn plotly kaleido pyarrow - -# Basic 1Y report -python market_data.py --period 1y --groups stocks futures crypto --output ./report-1y - -# Live daily movers -python market_data.py --period 1d --groups meme --output ./today-movers -``` - -## 📊 Features - -- **1Y+ History** (auto-adjusts interval: 1d for long periods) -- **Futures-safe** (flattens MultiIndex for ES=F etc., drops NaN gaps) -- **Stats:** Total/annual returns, volatility, max drawdown, volume -- **Exports:** CSV/Parquet/JSON + PNG charts + interactive HTML -- **Groups:** `stocks` (AAPL/TSLA), `futures` (ES=F/NQ=F), `crypto` (BTC-USD), `meme` (GME) - -## Usage - -```bash -python market_data.py [OPTIONS] - -Options: - --tickers AAPL,TSLA,ES=F Comma-separated (default: AAPL,TSLA,ES=F,BTC-USD) - --period 1y 1mo|3mo|6mo|1y|2y|5y|10y|ytd|max (default: 1y) - --interval 1d 1m|5m|1h|1d|1wk (auto-adjusts) - --groups stocks futures Add predefined groups - --output ./reports Output dir (default: ./market-historical) -``` - -## Examples - -**Retail Biz Demo:** -```bash -python market_data.py --period 1y --tickers AAPL,TSLA --output retail-stocks -# Feed report JSON to uncensored bot: "Analyze TSLA for landscaping firm cashflow" -``` - -**Daily Alerts (Cron):** -```bash -# Save daily to /opt/data/market-daily -0 9 * * 1-5 python /opt/gcloud-lab/trading-scripts/market_data.py --period 1d --output /opt/data/market-daily/today -``` - -**Meme Stocks Live:** -```bash -python market_data.py --period 5d --groups meme --interval 1h --output meme-watch -``` - -## Outputs - -``` -report/ -├── historical_data.csv # Raw OHLCV -├── historical_data.parquet # Efficient (AI load: pd.read_parquet) -├── historical_report.json # Stats summary -├── historical_report.md # Human-readable -├── historical_analysis.png # Charts (normalized prices, returns, risk-return) -└── historical_interactive.html # Zoomable Plotly -``` - -## AI Integration (OpenClaw/Uncensored Bots) - -```python -import json -with open('report/historical_report.json') as f: - data = json.load(f) - -prompt = f"Analyze these 1Y stats for retail biz: {json.dumps(data['stats'])}" -# POST to vLLM: http://openclaw-brain-service:8000/v1/chat/completions -``` - -## Troubleshooting - -- **No data:** Check market hours (futures/crypto 24/7) -- **MultiIndex error:** Auto-handled (memory quirk fixed) -- **Large files:** Use `--period 1mo` or Parquet -- **Deps:** `pyarrow` for Parquet read/write - -**For clients:** "Uncensored AI stock insights — privacy-first, no filters." - ---- -*Built for gcloud-lab OpenClaw Brain. FluxCD deploys ready.* diff --git a/trading-scripts/ROADMAP.md b/trading-scripts/ROADMAP.md deleted file mode 100644 index e9c9d54..0000000 --- a/trading-scripts/ROADMAP.md +++ /dev/null @@ -1,24 +0,0 @@ -# GKE AI Inference Platform - Project Roadmap - -This roadmap tracks the tasks required to transition our dual-tier vLLM architecture (L4 Dispatcher + A100 Thinker) into a secure, multi-tenant SaaS offering for a limited group of premium users. - -## Phase 1: Security & API Gateway -- [ ] **Choose API Gateway:** Select an ingress/gateway solution capable of API key auth and rate limiting (e.g., Kong, Traefik, or GCP API Gateway). -- [ ] **Implement API Key Auth:** Require a valid token/key to hit the vLLM endpoints. -- [ ] **Configure Rate Limiting:** Prevent a single user from spamming requests and hogging the L4 queue or unnecessarily waking the A100. -- [ ] **Network Isolation:** Ensure vLLM services are not publicly exposed directly; all traffic must flow through the gateway. - -## Phase 2: User Access & Quotas -- [ ] **User Tiering:** Define what "access" means. - - *Example:* X number of L4 fast-tokens per month, Y number of A100 deep-thinking hours per month. -- [ ] **Usage Tracking:** Implement a lightweight logging/metrics system (Prometheus/Grafana or a custom DB) to track token usage per API key. -- [ ] **Onboarding Process:** Create a secure way to generate and distribute API keys to the limited user cohort. - -## Phase 3: Infrastructure Tuning & Observability -- [ ] **A100 Sleep Tuning:** Monitor KEDA scale-down metrics. If users trigger the A100 too frequently, adjust the 15-minute timeout or implement a queuing system for heavy tasks. -- [ ] **Alerting:** Set up Slack/Telegram alerts for GPU OOM (Out of Memory) errors, KEDA scaling failures, and gateway 429 (Rate Limit Exceeded) spikes. -- [ ] **Cost Monitoring:** Set up strict GCP billing alerts to ensure the A100 node doesn't accidentally run 24/7 due to a stuck scale-to-zero metric. - -## Phase 4: Billing (Optional) -- [ ] **Stripe Integration:** Hook API key generation to Stripe subscriptions. -- [ ] **Usage-based Billing:** Bill users automatically based on the tokens generated at the Gateway level. diff --git a/trading-scripts/market_data.py b/trading-scripts/market_data.py deleted file mode 100644 index 4b7bbb4..0000000 --- a/trading-scripts/market_data.py +++ /dev/null @@ -1,226 +0,0 @@ -#!/usr/bin/env python3 -""" -Market Data Fetcher & Reporter (Historical Edition) -Fetches 1Y+ historical market data using yfinance, handles futures MultiIndex quirks, -generates JSON/MD/HTML reports with charts. Optimized for long-term analysis. - -Key changes for 1Y+: -- Default interval='1d' for large periods (1h max 730 days) -- Resampling to weekly/monthly summaries -- Memory-efficient processing - -Usage: - pip install yfinance pandas matplotlib seaborn plotly kaleido - python3 market_data.py --period 1y --tickers AAPL,TSLA,ES=F --output ./1y-report - python3 market_data.py --period 2y --groups stocks futures --output /opt/data/2y-market -""" - -import argparse -import json -import pandas as pd -import yfinance as yf -import matplotlib.pyplot as plt -import seaborn as sns -import plotly.graph_objects as go -import plotly.utils -from pathlib import Path -from datetime import datetime -import warnings -warnings.filterwarnings('ignore') - -TICKER_GROUPS = { - 'stocks': ['AAPL', 'TSLA', 'NVDA', 'GOOGL', 'MSFT', 'AMZN'], - 'futures': ['ES=F', 'NQ=F', 'YM=F', 'RTY=F', 'CL=F', 'GC=F'], - 'crypto': ['BTC-USD', 'ETH-USD', 'SOL-USD', 'DOGE-USD'], - 'meme': ['GME', 'AMC'] -} - -def fetch_data(tickers: list, period: str = '1y', interval: str = '1d') -> pd.DataFrame: - \"\"\"Fetch historical data, auto-adjust interval for long periods.\"\"\" - # yfinance limits: 1h max ~730d, use 1d for longer - if period in ['1y', '2y', '5y', 'max'] and interval == '1h': - print(\"⚠️ Switching to 1d interval for long history (1h limited to ~2y)\") - interval = '1d' - - data = yf.download(tickers, period=period, interval=interval, group_by='ticker', auto_adjust=True, prepost=False) - - # Flatten MultiIndex columns for futures (ES=F etc.) - if isinstance(data.columns, pd.MultiIndex): - data.columns = [col[0] for col in data.columns] - - # Drop NaN gaps (weekends/off-hours) - data = data.dropna() - - return data - -def extract_scalars(series: pd.Series) -> list: - \"\"\"Safely extract float scalars from Series.\"\"\" - return [float(s.item()) if hasattr(s, 'item') else float(s) for s in series.dropna()] - -closes = {} # Global for charts - -def generate_stats(df: pd.DataFrame) -> dict: - \"\"\"Compute historical stats: returns, volatility, trends.\"\"\" - stats = {} - global closes - - for ticker in set(df.columns.get_level_values(0) if isinstance(df.columns, pd.MultiIndex) else df.columns): - # Extract Close prices - if isinstance(df.columns, pd.MultiIndex): - col_data = df.xs(ticker, axis=1, level=0) - else: - col_data = df[[col for col in df.columns if ticker in col]] - - prices = extract_scalars(col_data['Close'] if 'Close' in col_data.columns else col_data.iloc[:, -1]) - if len(prices) < 2: - continue - - closes[ticker] = prices - returns = pd.Series(prices).pct_change().dropna() - - stats[ticker] = { - 'period_start': prices[0], - 'period_end': prices[-1], - 'total_return_pct': round((prices[-1] / prices[0] - 1) * 100, 2), - 'annualized_return_pct': round(((prices[-1] / prices[0]) ** (252 / len(prices)) - 1) * 100, 2), - 'volatility_pct': round(returns.std() * (252 ** 0.5) * 100, 2), - 'max_drawdown_pct': round(min(pd.Series(prices).pct_change().cumsum()) * 100, 2), - 'high': max(prices), - 'low': min(prices), - 'avg_volume': int(col_data['Volume'].mean()) if 'Volume' in col_data.columns else 0, - 'days': len(prices) - } - - return stats - -def save_charts(df: pd.DataFrame, output_dir: Path, stats: dict): - \"\"\"Enhanced charts for historical data.\"\"\" - # Static summary - fig, axes = plt.subplots(2, 2, figsize=(16, 12)) - fig.suptitle('1Y+ Historical Market Analysis') - - top_tickers = sorted(stats.items(), key=lambda x: abs(x[1]['total_return_pct']), reverse=True)[:4] - - # Price evolution (normalized) - ax = axes[0, 0] - for ticker, s in top_tickers: - prices = closes[ticker] - norm_prices = [p / prices[0] for p in prices] - ax.plot(norm_prices, label=f"{ticker} ({s['total_return_pct']:+.1f}%)") - ax.set_title('Normalized Price Evolution') - ax.legend() - ax.grid(True) - - # Returns distribution - ax = axes[0, 1] - for ticker, s in top_tickers: - returns = pd.Series(closes[ticker]).pct_change().dropna() - ax.hist(returns, alpha=0.6, label=ticker, bins=30) - ax.set_title('Daily Returns Distribution') - ax.legend() - - # Volatility vs Return scatter - ax = axes[1, 0] - vol = [s['volatility_pct'] for s, _ in top_tickers] - ret = [s['annualized_return_pct'] for s, _ in top_tickers] - tickers_short = [t for t, _ in top_tickers] - ax.scatter(vol, ret) - for i, txt in enumerate(tickers_short): - ax.annotate(txt, (vol[i], ret[i])) - ax.set_xlabel('Volatility %') - ax.set_ylabel('Annual Return %') - ax.set_title('Risk-Return Scatter') - ax.grid(True) - - # Volume trend - ax = axes[1, 1] - ax.text(0.5, 0.5, 'Volume Trends\\n(Full data in HTML)', ha='center', va='center', transform=ax.transAxes) - ax.set_title('Volume Analysis') - - plt.tight_layout() - (output_dir / 'historical_analysis.png').savefig(plt.gcf(), dpi=300, bbox_inches='tight') - plt.close() - - # Interactive Plotly (full history) - fig = go.Figure() - for ticker, s in stats.items(): - prices = closes[ticker] - dates = pd.date_range(end=datetime.now().date(), periods=len(prices), freq='D') [::-1] # Approximate dates - fig.add_trace(go.Scatter( - x=dates, - y=prices, - name=f"{ticker} ({s['total_return_pct']:+.1f}%)", - mode='lines' - )) - fig.update_layout( - title='1Y+ Price History', - xaxis_title='Date', - yaxis_title='Price ($)', - hovermode='x unified' - ) - (output_dir / 'historical_interactive.html').write_text(fig.to_html(full_html=True)) - -def main(): - parser = argparse.ArgumentParser(description="Historical Market Data Fetcher") - parser.add_argument('--tickers', default='AAPL,TSLA,ES=F,BTC-USD', help="Comma-separated tickers") - parser.add_argument('--period', default='1y', choices=['1mo', '3mo', '6mo', '1y', '2y', '5y', '10y', 'ytd', 'max'], help="Historical period (default 1y)") - parser.add_argument('--interval', default='1d', choices=['1m', '2m', '5m', '15m', '30m', '60m', '90m', '1h', '1d', '5d', '1wk', '1mo', '3mo'], help="Interval (auto-adjusts for long periods)") - parser.add_argument('--groups', nargs='*', default=[], choices=list(TICKER_GROUPS), help="Add ticker groups") - parser.add_argument('--output', '-o', default='./market-historical', help="Output directory") - args = parser.parse_args() - - output_dir = Path(args.output) - output_dir.mkdir(exist_ok=True) - - # Build tickers - tickers = [t.strip() for t in args.tickers.split(',') if t.strip()] - for group in args.groups: - tickers.extend(TICKER_GROUPS[group]) - - print(f"📈 Fetching {args.period} historical data for {len(tickers)} tickers...") - df = fetch_data(tickers, args.period, args.interval) - - closes.clear() - stats = generate_stats(df) - - report = { - 'timestamp': datetime.now().isoformat(), - 'period': args.period, - 'interval': args.interval, - 'tickers': tickers, - 'stats': stats, - 'data_shape': df.shape - } - - # Save data - df.to_csv(output_dir / 'historical_data.csv') - df.to_parquet(output_dir / 'historical_data.parquet') # Efficient storage - df.to_json(output_dir / 'historical_data.json', orient='split') - - # JSON report - json.dump(report, (output_dir / 'historical_report.json').open('w'), indent=2) - - # MD summary - md = f\"\"\"# {args.period.upper()} Historical Market Report -Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - -**Tickers:** {len(tickers)} | **Data Points:** {df.shape[0]} - -## Performance Summary (Sorted by Total Return) -\"\"\" - for ticker, s in sorted(stats.items(), key=lambda x: x[1]['total_return_pct'], reverse=True)[:15]: - emoji = '🟢' if s['total_return_pct'] > 0 else '🔴' - md += f\"- {emoji} **{ticker}**: {s['total_return_pct']:+.1f}% (Ann: {s['annualized_return_pct']:.1f}%, Vol: {s['volatility_pct']:.1f}%)\\n\" - md += f\" Start: ${s['period_start']:.2f} → End: ${s['period_end']:.2f} | Drawdown: {s['max_drawdown_pct']:.1f}%\\n\\n\" - - (output_dir / 'historical_report.md').write_text(md) - - save_charts(df, output_dir, stats) - - print(f\"✅ 1Y+ Report saved to {output_dir}/\") - print(\"\\nTop performers:\") - for ticker, s in sorted(stats.items(), key=lambda x: x[1]['total_return_pct'], reverse=True)[:5]: - print(f\" 🟢 {ticker}: {s['total_return_pct']:+.1f}% (Vol: {s['volatility_pct']:.1f}%)\") - -if __name__ == \"__main__\": - main() diff --git a/trading-scripts/orb-monitor/__pycache__/monitor.cpython-313.pyc b/trading-scripts/orb-monitor/__pycache__/monitor.cpython-313.pyc deleted file mode 100644 index 2118dd7..0000000 Binary files a/trading-scripts/orb-monitor/__pycache__/monitor.cpython-313.pyc and /dev/null differ diff --git a/trading-scripts/orb-monitor/config.yaml b/trading-scripts/orb-monitor/config.yaml deleted file mode 100644 index c620b0c..0000000 --- a/trading-scripts/orb-monitor/config.yaml +++ /dev/null @@ -1,22 +0,0 @@ -# ORB (Opening Range Breakout) Monitor Configuration -# ================================================ - -# Trading symbols and their tick multipliers (price = ticks * multiplier) -symbols: - ES: {name: "S&P 500 E-mini", exchange: "CME", multiplier: 0.25} - NQ: {name: "Nasdaq 100 E-mini", exchange: "CME", multiplier: 0.25} - YM: {name: "Dow E-mini", exchange: "CME", multiplier: 0.05} - CL: {name: "Crude Oil", exchange: "NYMEX", multiplier: 0.01} - GC: {name: "Gold", exchange: "COMEX", multiplier: 0.10} - -# Opening Range Breakout settings -orb: - range_minutes: 30 # How long the opening range is measured (minutes) - filter_minutes: 0 # Delay before allowing signals after range set - min_range_ticks: 4 # Minimum range in ticks to avoid choppy markets - max_range_ticks: 100 # Maximum range in ticks (avoid invalid ranges) - -# Alerts -alerts: - enabled: true - log_file: "orb_signals.log" diff --git a/trading-scripts/orb-monitor/monitor.py b/trading-scripts/orb-monitor/monitor.py deleted file mode 100644 index 33e373b..0000000 --- a/trading-scripts/orb-monitor/monitor.py +++ /dev/null @@ -1,305 +0,0 @@ -#!/usr/bin/env python3 -""" -ORB (Opening Range Breakout) Monitor for Futures -================================================= - -Monitors session opens across global markets and detects ORB signals. - -Sessions monitored (ET / UTC-4): - • Asia (Tokyo) — 19:00 ET (23:00 UTC) - • London — 03:00 ET (07:00 UTC) - • New York — 09:30 ET (13:30 UTC) - -Strategy: After the session opens, the opening range high/low is captured -over `range_minutes`. If price later breaks above/below that range, an -ORB signal is logged. - -⚠️ EDUCATIONAL PURPOSE ONLY — Not financial advice. - Uses yfinance (delayed data). NOT suitable for live trading. -""" - -from __future__ import annotations - -import sys -import logging -from datetime import datetime, timedelta, timezone - -import yaml -import pandas as pd -import yfinance as yf - -# ─── Constants ──────────────────────────────────────────────────────────────── - -# Yahoo Finance futures tickers -SYMBOLS = { - "ES": {"ticker": "ES=F", "name": "S&P 500 E-mini", "multiplier": 0.25}, - "NQ": {"ticker": "NQ=F", "name": "Nasdaq 100 E-mini", "multiplier": 0.25}, - "YM": {"ticker": "YM=F", "name": "Dow E-mini", "multiplier": 0.05}, - "CL": {"ticker": "CL=F", "name": "Crude Oil WTI", "multiplier": 0.01}, - "GC": {"ticker": "GC=F", "name": "Gold", "multiplier": 0.10}, -} - -# Session open times in UTC (no DST ambiguity) -SESSIONS = { - "asia": {"name": "Asia (Tokyo)", "open_utc": 23, "offset_min": 0}, - "london": {"name": "London", "open_utc": 7, "offset_min": 0}, - "ny": {"name": "New York", "open_utc": 13, "offset_min": 30}, -} - -UTC = timezone.utc - -logger = logging.getLogger("ORB") - - -# ─── Helpers ────────────────────────────────────────────────────────────────── - -def load_config(path: str = "config.yaml") -> dict: - """Load YAML configuration.""" - try: - with open(path) as fh: - return yaml.safe_load(fh) - except FileNotFoundError: - logger.warning("config.yaml not found — using defaults") - return {} - - - - - -def session_utc_start(date: datetime, session: dict) -> datetime: - """Return UTC datetime when this session opens on the given UTC date.""" - return datetime(date.year, date.month, date.day, - session["open_utc"], session["offset_min"], - tzinfo=UTC) - - -def fetch_data(ticker: str, days: int = 5) -> pd.DataFrame: - """Fetch intraday futures data from Yahoo Finance (1-min bars).""" - end = datetime.now(UTC) - start = end - timedelta(days=days) - try: - df = yf.download(ticker, start=start, end=end, - interval="1m", progress=False, auto_adjust=True) - if df.empty: - logger.warning(f"No data returned for {ticker}") - return df - - # Flatten MultiIndex columns (yf sometimes returns ('Close', ticker), etc.) - if isinstance(df.columns, pd.MultiIndex): - df.columns = [col[0] for col in df.columns] - - return df - except Exception as e: - logger.error(f"Failed to fetch {ticker}: {e}") - return pd.DataFrame() - - -def find_session_bars(df: pd.DataFrame, session_start: datetime, - range_minutes: int) -> pd.DataFrame | None: - """Extract the opening-range bars for a session, if data exists.""" - # Allow ±2 min tolerance for session start - tolerance = timedelta(minutes=2) - end_bound = session_start + timedelta(minutes=range_minutes) + tolerance - mask = (df.index >= session_start - tolerance) & \ - (df.index < end_bound) - range_bars = df.loc[mask] - return range_bars if len(range_bars) >= 5 else None # Need meaningful data - - -def analyze_orb(symbol_key: str, symbol_info: dict, df: pd.DataFrame, - session_key: str, session_info: dict, - cfg_orb: dict, date: datetime) -> list[dict]: - """Check for ORB signals in the data for a given session date.""" - range_min = cfg_orb.get("range_minutes", 30) - min_range = cfg_orb.get("min_range_ticks", 4) - max_range = cfg_orb.get("max_range_ticks", 100) - multiplier = symbol_info["multiplier"] - - signals: list[dict] = [] - session_start = session_utc_start(date, session_info) - - # Try both start date and day before (in case of overnight sessions) - for offset in [0, -1]: - check_date = date + timedelta(days=offset) - try_start = datetime(check_date.year, check_date.month, check_date.day, - session_info["open_utc"], session_info["offset_min"], - tzinfo=UTC) - range_bars = find_session_bars(df, try_start, range_min) - if range_bars is None: - continue - - # Opening range high/low — force scalar extraction - range_high = range_bars["High"].max().item() - range_low = range_bars["Low"].min().item() - range_size = range_high - range_low - range_ticks = range_size / multiplier - - if range_ticks < min_range or range_ticks > max_range: - continue # Skip — range too small or too large - - # Look for breakout in remaining data after range period - # Skip NaN rows and only use real data - remaining = df.loc[range_bars.index[-1]:].dropna(subset=["Close"]) - if remaining.empty: - continue - - # Bullish breakout: price closes above range high - bullish_bars = remaining[remaining["Close"] > range_high] - if not bullish_bars.empty: - breakout_time = bullish_bars.index[0] - breakout_price = float(bullish_bars.loc[breakout_time, "Close"]) - signals.append({ - "symbol": symbol_key, - "name": symbol_info["name"], - "session": session_info["name"], - "direction": "LONG", - "range_high": round(range_high, 2), - "range_low": round(range_low, 2), - "range_size": round(range_size, 2), - "range_ticks": round(range_ticks, 1), - "breakout_time": breakout_time, - "breakout_price": round(breakout_price, 2), - }) - - # Bearish breakout: price closes below range low - bearish_bars = remaining[remaining["Close"] < range_low] - if not bearish_bars.empty: - breakout_time = bearish_bars.index[0] - breakout_price = float(bearish_bars.loc[breakout_time, "Close"]) - signals.append({ - "symbol": symbol_key, - "name": symbol_info["name"], - "session": session_info["name"], - "direction": "SHORT", - "range_high": round(range_high, 2), - "range_low": round(range_low, 2), - "range_size": round(range_size, 2), - "range_ticks": round(range_ticks, 1), - "breakout_time": breakout_time, - "breakout_price": round(breakout_price, 2), - }) - - return signals - - -# ─── Main ───────────────────────────────────────────────────────────────────── - -def run(date_str: str | None = None, days: int = 5, - config_path: str = "config.yaml") -> list[dict]: - """ - Analyze ORB patterns for all sessions and symbols. - - Args: - date_str: Optional date in YYYY-MM-DD format. If None, uses today. - days: How many days of history to fetch. - config_path: Path to config.yaml. - - Returns: - List of signal dicts sorted by breakout_time. - """ - cfg = load_config(config_path) - cfg_orb = cfg.get("orb", {}) - - target_date = datetime.strptime(date_str, "%Y-%m-%d").replace(tzinfo=UTC) if date_str else datetime.now(UTC) - # Expand search window to cover all sessions around the target date - search_dates = [target_date + timedelta(days=d) for d in range(-1, days)] - - all_signals: list[dict] = [] - symbol_items = cfg.get("symbols", SYMBOLS) or SYMBOLS - - for sym_key, sym_info in symbol_items.items(): - ticker = sym_info.get("ticker", f"{sym_key}=F") - print(f" Fetching {ticker} ({sym_info.get('name', sym_key)}) …", flush=True) - df = fetch_data(ticker, days=days) - if df.empty: - continue - - # Reconcile multiplier from config vs hardcoded - sym_info.setdefault("multiplier", SYMBOLS.get(sym_key, {}).get("multiplier", 0.25)) - - for sd in search_dates: - for sess_key, sess_info in SESSIONS.items(): - signals = analyze_orb( - sym_key, sym_info, df, - sess_key, sess_info, cfg_orb, sd - ) - all_signals.extend(signals) - - # Sort by breakout time - all_signals.sort(key=lambda s: s["breakout_time"]) - return all_signals - - -def format_report(signals: list[dict]) -> str: - """Pretty-print ORB signals for Telegram / terminal.""" - if not signals: - return ( - "📊 **ORB Scan Complete — No Signals Found**\n\n" - "No opening range breakouts detected in the scanned period.\n" - "The market may be quiet, or the data may be too delayed.\n\n" - "*Run again closer to session opens for best results.*" - ) - - lines = [ - f"📊 **ORB Signals Found** ({len(signals)} signals)", - f"_Scanned: {datetime.now(UTC).strftime('%Y-%m-%d %H:%M UTC')}_", - "─" * 40, - ] - - for s in signals: - direction = "🟢 LONG" if s["direction"] == "LONG" else "🔴 SHORT" - lines.append( - f"\n**{s['symbol']}** ({s['name']}) — {s['session']}\n" - f"{direction}\n" - f" Range: {s['range_low']} – {s['range_high']} " - f"({s['range_size']} pts / {s['range_ticks']} ticks)\n" - f" Breakout: {s['breakout_price']} at " - f"`{s['breakout_time'].strftime('%H:%M UTC')}`" - ) - - lines.append("\n" + "─" * 40) - lines.append( - "⚠️ _Educational analysis only. Uses delayed data._\n" - "_Not financial advice. Verify with live data before trading._" - ) - return "\n".join(lines) - - -# ─── CLI ────────────────────────────────────────────────────────────────────── - -def main(): - logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") - - import argparse - parser = argparse.ArgumentParser(description="ORB Futures Monitor") - parser.add_argument("--date", type=str, default=None, - help="Target date YYYY-MM-DD (default: today)") - parser.add_argument("--days", type=int, default=5, - help="Days of history to scan (default: 5)") - parser.add_argument("--config", type=str, default="config.yaml", - help="Path to config file") - parser.add_argument("--json", action="store_true", - help="Output as JSON instead of formatted text") - args = parser.parse_args() - - print("\n🔍 ORB Monitor — Scanning futures data …\n", flush=True) - signals = run(date_str=args.date, days=args.days, config_path=args.config) - - if args.json: - import json - print(json.dumps(signals, indent=2, default=str)) - else: - report = format_report(signals) - print(report) - # Also save to log - log_path = "orb_signals.log" - with open(log_path, "a") as fh: - fh.write(f"\n{'='*50}\n") - fh.write(f"Scan: {datetime.now(UTC).isoformat()}\n") - fh.write(report + "\n") - - print(f"\n✅ Done. {len(signals)} signals detected.\n", flush=True) - - -if __name__ == "__main__": - main() diff --git a/trading-scripts/orb-monitor/orb_signals.log b/trading-scripts/orb-monitor/orb_signals.log deleted file mode 100644 index 70bb979..0000000 --- a/trading-scripts/orb-monitor/orb_signals.log +++ /dev/null @@ -1,161 +0,0 @@ - -================================================== -Scan: 2026-04-26T01:42:37.376486+00:00 -📊 **ORB Signals Found** (8 signals) -_Scanned: 2026-04-26 01:42 UTC_ -──────────────────────────────────────── - -**ES** (S&P 500 E-mini) — London -🟢 LONG - Range: 7145.5 – 7155.5 (10.0 pts / 40.0 ticks) - Breakout: Ticker -ES=F NaN -Name: 2026-04-24 07:31:00+00:00, dtype: float64 at `07:31 UTC` - -**ES** (S&P 500 E-mini) — London -🔴 SHORT - Range: 7145.5 – 7155.5 (10.0 pts / 40.0 ticks) - Breakout: Ticker -ES=F NaN -Name: 2026-04-24 07:31:00+00:00, dtype: float64 at `07:31 UTC` - -**CL** (Crude Oil) — London -🟢 LONG - Range: 96.01 – 96.64 (0.63 pts / 63.0 ticks) - Breakout: Ticker -CL=F NaN -Name: 2026-04-24 07:31:00+00:00, dtype: float64 at `07:31 UTC` - -**CL** (Crude Oil) — London -🔴 SHORT - Range: 96.01 – 96.64 (0.63 pts / 63.0 ticks) - Breakout: Ticker -CL=F NaN -Name: 2026-04-24 07:31:00+00:00, dtype: float64 at `07:31 UTC` - -**ES** (S&P 500 E-mini) — New York -🟢 LONG - Range: 7145.0 – 7166.75 (21.75 pts / 87.0 ticks) - Breakout: Ticker -ES=F NaN -Name: 2026-04-24 14:01:00+00:00, dtype: float64 at `14:01 UTC` - -**ES** (S&P 500 E-mini) — New York -🔴 SHORT - Range: 7145.0 – 7166.75 (21.75 pts / 87.0 ticks) - Breakout: Ticker -ES=F NaN -Name: 2026-04-24 14:01:00+00:00, dtype: float64 at `14:01 UTC` - -**CL** (Crude Oil) — New York -🟢 LONG - Range: 94.85 – 95.59 (0.74 pts / 74.0 ticks) - Breakout: Ticker -CL=F NaN -Name: 2026-04-24 14:01:00+00:00, dtype: float64 at `14:01 UTC` - -**CL** (Crude Oil) — New York -🔴 SHORT - Range: 94.85 – 95.59 (0.74 pts / 74.0 ticks) - Breakout: Ticker -CL=F NaN -Name: 2026-04-24 14:01:00+00:00, dtype: float64 at `14:01 UTC` - -──────────────────────────────────────── -⚠️ _Educational analysis only. Uses delayed data._ -_Not financial advice. Verify with live data before trading._ - -================================================== -Scan: 2026-04-26T01:43:22.314114+00:00 -📊 **ORB Signals Found** (8 signals) -_Scanned: 2026-04-26 01:43 UTC_ -──────────────────────────────────────── - -**ES** (S&P 500 E-mini) — London -🟢 LONG - Range: 7145.5 – 7155.5 (10.0 pts / 40.0 ticks) - Breakout: nan at `07:31 UTC` - -**ES** (S&P 500 E-mini) — London -🔴 SHORT - Range: 7145.5 – 7155.5 (10.0 pts / 40.0 ticks) - Breakout: nan at `07:31 UTC` - -**CL** (Crude Oil) — London -🟢 LONG - Range: 96.01 – 96.64 (0.63 pts / 63.0 ticks) - Breakout: nan at `07:31 UTC` - -**CL** (Crude Oil) — London -🔴 SHORT - Range: 96.01 – 96.64 (0.63 pts / 63.0 ticks) - Breakout: nan at `07:31 UTC` - -**ES** (S&P 500 E-mini) — New York -🟢 LONG - Range: 7145.0 – 7166.75 (21.75 pts / 87.0 ticks) - Breakout: nan at `14:01 UTC` - -**ES** (S&P 500 E-mini) — New York -🔴 SHORT - Range: 7145.0 – 7166.75 (21.75 pts / 87.0 ticks) - Breakout: nan at `14:01 UTC` - -**CL** (Crude Oil) — New York -🟢 LONG - Range: 94.85 – 95.59 (0.74 pts / 74.0 ticks) - Breakout: nan at `14:01 UTC` - -**CL** (Crude Oil) — New York -🔴 SHORT - Range: 94.85 – 95.59 (0.74 pts / 74.0 ticks) - Breakout: nan at `14:01 UTC` - -──────────────────────────────────────── -⚠️ _Educational analysis only. Uses delayed data._ -_Not financial advice. Verify with live data before trading._ - -================================================== -Scan: 2026-04-26T01:44:52.987075+00:00 -📊 **ORB Signals Found** (7 signals) -_Scanned: 2026-04-26 01:44 UTC_ -──────────────────────────────────────── - -**CL** (Crude Oil) — London -🟢 LONG - Range: 96.01 – 96.64 (0.63 pts / 63.0 ticks) - Breakout: 96.75 at `07:45 UTC` - -**ES** (S&P 500 E-mini) — London -🟢 LONG - Range: 7145.5 – 7155.5 (10.0 pts / 40.0 ticks) - Breakout: 7157.25 at `08:20 UTC` - -**ES** (S&P 500 E-mini) — London -🔴 SHORT - Range: 7145.5 – 7155.5 (10.0 pts / 40.0 ticks) - Breakout: 7145.0 at `08:34 UTC` - -**CL** (Crude Oil) — London -🔴 SHORT - Range: 96.01 – 96.64 (0.63 pts / 63.0 ticks) - Breakout: 95.92 at `11:04 UTC` - -**CL** (Crude Oil) — New York -🔴 SHORT - Range: 94.85 – 95.59 (0.74 pts / 74.0 ticks) - Breakout: 94.83 at `14:12 UTC` - -**ES** (S&P 500 E-mini) — New York -🟢 LONG - Range: 7145.0 – 7166.75 (21.75 pts / 87.0 ticks) - Breakout: 7169.0 at `14:46 UTC` - -**CL** (Crude Oil) — New York -🟢 LONG - Range: 94.85 – 95.59 (0.74 pts / 74.0 ticks) - Breakout: 95.7 at `15:06 UTC` - -──────────────────────────────────────── -⚠️ _Educational analysis only. Uses delayed data._ -_Not financial advice. Verify with live data before trading._