From Zero
To Production
Price Action Scanner

import os
from SmartApi import SmartConnect
import pyotp
import pandas as pd
from datetime import datetime, timedelta
import logging

logger = logging.getLogger(__name__)

class AngelOneIngestor:
    """
    Handles all data fetching from Angel One SmartAPI.
    Supports historical OHLCV and live tick data.
    """

    TIMEFRAME_MAP = {
        "5m":  "FIVE_MINUTE",
        "15m": "FIFTEEN_MINUTE",
        "1h":  "ONE_HOUR",
        "1d":  "ONE_DAY",
    }

    def __init__(self):
        self.api = SmartConnect(api_key=os.getenv("ANGEL_API_KEY"))
        totp = pyotp.TOTP(os.getenv("ANGEL_TOTP_SECRET"))
        self.api.generateSession(
            os.getenv("ANGEL_CLIENT_ID"),
            os.getenv("ANGEL_MPIN"),
            totp.now()
        )
        logger.info("✅ Angel One session established")

    def fetch_candles(
        self,
        symbol_token: str,    # e.g. "2885" for RELIANCE
        exchange: str = "NSE",
        timeframe: str = "5m",
        lookback_days: int = 30
    ) -> pd.DataFrame:
        """Fetch OHLCV candles for a single symbol."""
        
        to_date   = datetime.now()
        from_date = to_date - timedelta(days=lookback_days)
        
        params = {
            "exchange":    exchange,
            "symboltoken": symbol_token,
            "interval":    self.TIMEFRAME_MAP[timeframe],
            "fromdate":    from_date.strftime("%Y-%m-%d %H:%M"),
            "todate":      to_date.strftime("%Y-%m-%d %H:%M"),
        }
        
        resp = self.api.getCandleData(params)
        if not resp["data"]:
            raise ValueError(f"No data returned for token {symbol_token}")
        
        df = pd.DataFrame(
            resp["data"],
            columns=["timestamp","open","high","low","close","volume"]
        )
        df["timestamp"] = pd.to_datetime(df["timestamp"])
        df = df.set_index("timestamp").sort_index()
        
        return self._clean(df)

    def _clean(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Preprocessing pipeline:
        1. Remove pre-market / post-market bars (keep 09:15 - 15:30)
        2. Drop rows with zero volume (non-trading bars)
        3. Forward-fill micro-gaps (<= 2 bars)
        4. Remove clear outliers (price > 5x median or < 0)
        """
        # 1. Market hours filter
        df = df.between_time("09:15", "15:30")
        
        # 2. Zero-volume drop
        df = df[df["volume"] > 0]
        
        # 3. Forward-fill gaps up to 2 bars
        full_idx = pd.date_range(
            df.index.min(), df.index.max(), freq="5min"
        )
        df = df.reindex(full_idx).ffill(limit=2).dropna()
        
        # 4. Outlier removal
        med_close = df["close"].median()
        df = df[(df["close"] > 0) & (df["close"] < med_close * 5)]
        
        return df.astype(float)

    def resample_to_15m(self, df_5m: pd.DataFrame) -> pd.DataFrame:
        """Convert 5-min bars to 15-min bars using OHLCV aggregation."""
        return df_5m.resample("15min").agg({
            "open":   "first",
            "high":   "max",
            "low":    "min",
            "close":  "last",
            "volume": "sum",
        }).dropna()

import pandas as pd
import numpy as np

class FeatureEngineer:
    """
    Builds feature vector for XGBoost inference.
    ⚠️  CRITICAL: all features use ONLY past data (df.iloc[:-1])
                  to prevent look-ahead bias.
    """

    def build(self, df: pd.DataFrame,
               signal, levels: dict) -> dict:
        """Build full feature dict for a given signal + OHLCV dataframe."""
        bar = df.iloc[-1]
        hist = df.iloc[:-1]  # ← ONLY HISTORICAL, never current bar

        features = {}

        # ── Price Action Features ──────────────────────────────
        bar_range = bar["high"] - bar["low"]
        features["body_pct"]  = abs(bar["close"] - bar["open"]) / (bar_range + 1e-9)
        features["wick_upper"]= (bar["high"] - max(bar["open"],bar["close"])) / (bar_range + 1e-9)
        features["wick_lower"]= (min(bar["open"],bar["close"]) - bar["low"]) / (bar_range + 1e-9)
        features["close_pct"] = (bar["close"] - bar["low"]) / (bar_range + 1e-9)
        prev = hist.iloc[-1]
        features["gap_pct"]   = (bar["open"] - prev["close"]) / (prev["close"] + 1e-9)
        features["ret_1bar"]  = (bar["close"] - prev["close"]) / (prev["close"] + 1e-9)

        # ── S/R Level Features ────────────────────────────────
        features["sr_touches"]   = signal.touches
        features["level_break_pct"] = abs(bar["close"] - signal.level) / signal.level
        features["liq_grab"]     = int(signal.liq_grab)

        # ── Volume Features ───────────────────────────────────
        vol_window = hist["volume"].iloc[-20:]
        features["vol_ratio_20"] = bar["volume"] / (vol_window.mean() + 1)
        features["vol_trend_5"]  = np.polyfit(
            range(5), hist["volume"].iloc[-5:].values, 1
        )[0]

        # ── Momentum Indicators (computed on hist only) ───────
        features["rsi_14"]  = self._rsi(hist, 14)
        features["adx_14"]  = self._adx(hist, 14)
        features["macd_hist"] = self._macd_hist(hist)

        # ── Volatility ────────────────────────────────────────
        atr = self._atr(hist, 14)
        features["atr_pct"]  = atr / bar["close"]
        log_rets = np.log(hist["close"] / hist["close"].shift(1)).dropna()
        features["hist_vol_10"] = log_rets.iloc[-10:].std()

        # ── Time Context ──────────────────────────────────────
        ts = df.index[-1]
        features["hour_of_day"]   = ts.hour + ts.minute / 60
        open_time = ts.replace(hour=9, minute=15)
        features["mins_from_open"] = (int)((ts - open_time).total_seconds() / 60)
        features["is_first_30min"] = int(features["mins_from_open"] <= 30)
        features["is_last_30min"]  = int(features["mins_from_open"] >= 330)

        return features

    def _rsi(self, df, p=14):
        d = df["close"].diff()
        g = d.clip(lower=0).ewm(alpha=1/p).mean()
        l = (-d).clip(lower=0).ewm(alpha=1/p).mean()
        rs = g / (l + 1e-9)
        return (100 - 100 / (1 + rs)).iloc[-1]

    # _adx and _macd_hist follow same pattern (omitted for brevity)
    def _atr(self, df, p=14):
        h, l, pc = df["high"], df["low"], df["close"].shift(1)
        tr = pd.concat([h-l,(h-pc).abs(),(l-pc).abs()],axis=1).max(axis=1)
        return tr.rolling(p).mean().iloc[-1]

import schedule, time, logging
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor
import xgboost as xgb

from data.ingestor import AngelOneIngestor
from patterns.sr_detector import SRDetector
from patterns.breakout import BreakoutEngine
from ml.features import FeatureEngineer
from alerts.telegram_bot import TelegramAlerter
from scanner.watchlist import NIFTY500_TOKENS   # {symbol: token} dict

logger = logging.getLogger(__name__)

class Scanner5m:
    """
    5-Minute Intraday Breakout/Breakdown Scanner.
    Runs on every 5-min bar close (09:20, 09:25, ..., 15:25).
    Scans all 500 stocks in parallel using thread pool.
    """

    MIN_ML_SCORE = 0.60  # only alert if model confidence ≥ 60%
    WORKERS      = 20    # parallel threads for 500 stocks

    def __init__(self):
        self.ingestor  = AngelOneIngestor()
        self.sr        = SRDetector(min_touches=2)
        self.engine    = BreakoutEngine()
        self.features  = FeatureEngineer()
        self.alerter   = TelegramAlerter()
        self.model     = xgb.Booster()
        self.model.load_model("ml/model_5m.json")
        self.alerted_today = set()  # avoid duplicate alerts

    def scan_symbol(self, symbol: str, token: str):
        """Pipeline for a single stock. Returns Signal or None."""
        try:
            df     = self.ingestor.fetch_candles(token, timeframe="5m")
            levels = self.sr.find_levels(df.iloc[:-1])  # ← no look-ahead!
            signal = self.engine.evaluate(symbol, df, levels, "5m")
            if signal is None:
                return None
            
            feats   = self.features.build(df, signal, levels)
            dmat    = xgb.DMatrix([list(feats.values())])
            score   = float(self.model.predict(dmat)[0])
            signal.ml_score = score

            if score >= self.MIN_ML_SCORE:
                logger.info(f"🎯 {symbol} score={score:.2f} → ALERT")
                return signal
        except Exception as e:
            logger.warning(f"{symbol} scan failed: {e}")
        return None

    def run_scan(self):
        """Run full 500-stock scan at bar close. Called by scheduler."""
        now = datetime.now()
        logger.info(f"⏱ Scan started at {now.strftime('%H:%M:%S')}")
        
        items = list(NIFTY500_TOKENS.items())
        signals_found = []

        with ThreadPoolExecutor(max_workers=self.WORKERS) as ex:
            futures = {
                ex.submit(self.scan_symbol, sym, tok): sym
                for sym, tok in items
                if sym not in self.alerted_today
            }
            for f in futures:
                result = f.result()
                if result:
                    signals_found.append(result)
                    self.alerted_today.add(result.symbol)

        # Sort by ML score, send top alerts
        signals_found.sort(key=lambda s: s.ml_score, reverse=True)
        for sig in signals_found[:10]:  # max 10 alerts per scan
            self.alerter.send(sig, sig.ml_score)
        
        logger.info(f"✅ Scan done. Found {len(signals_found)} signals.")

    def start(self):
        """Schedule scans at every 5-min bar close during market hours."""
        # Run at :20 past every hour (to ensure 5-min bar is fully closed)
        for minute in [20,25,30,35,40,45,50,55]:
            schedule.every().hour.at(f":{minute:02d}").do(self.run_scan)
        schedule.every().hour.at(":00").do(self.run_scan)
        schedule.every().hour.at(":05").do(self.run_scan)
        schedule.every().hour.at(":10").do(self.run_scan)
        schedule.every().hour.at(":15").do(self.run_scan)

        logger.info("🚀 5-min scanner running. Ctrl+C to stop.")
        while True:
            schedule.run_pending()
            time.sleep(1)

if __name__ == "__main__":
    scanner = Scanner5m()
    scanner.start()

from fastapi import FastAPI, WebSocket
from fastapi.middleware.cors import CORSMiddleware
import json

app = FastAPI(title="Nifty500 Scanner API", version="1.0")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"])

# In-memory signal store (replace with Redis/DB in production)
active_signals = []

@app.get("/signals")
async def get_signals(
    timeframe: str = "all",
    min_score: float = 0.6,
    signal_type: str = "all"
):
    """Return filtered list of active signals."""
    results = active_signals
    if timeframe != "all":
        results = [s for s in results if s["timeframe"] == timeframe]
    if signal_type != "all":
        results = [s for s in results if s["type"] == signal_type]
    results = [s for s in results if s["ml_score"] >= min_score]
    return {"signals": sorted(results, key=lambda x: -x["ml_score"])}

@app.websocket("/ws/signals")
async def websocket_signals(ws: WebSocket):
    """WebSocket for real-time signal push to web UI."""
    await ws.accept()
    try:
        while True:
            await ws.send_json({"signals": active_signals[-20:]})
            await asyncio.sleep(5)
    except:
        pass

@app.get("/health")
async def health():
    return {"status": "ok", "active_signals": len(active_signals)}

version: "3.9"
services:

  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]
    restart: always

  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB:       scanner_db
      POSTGRES_USER:     scanner
      POSTGRES_PASSWORD: "${DB_PASSWORD}"
    volumes: ["pgdata:/var/lib/postgresql/data"]
    restart: always

  scanner:
    build: .
    command: python -m scanner.scanner_5m
    env_file: .env
    depends_on: [redis, postgres]
    restart: always

  scanner_15m:
    build: .
    command: python -m scanner.scanner_15m
    env_file: .env
    depends_on: [redis, postgres]
    restart: always

  api:
    build: .
    command: uvicorn api.main:app --host 0.0.0.0 --port 8000
    ports: ["8000:8000"]
    env_file: .env
    depends_on: [redis, postgres]
    restart: always

  frontend:
    image: nginx:alpine
    ports: ["80:80", "443:443"]
    volumes:
      - ./frontend/dist:/usr/share/nginx/html
      - ./nginx.conf:/etc/nginx/conf.d/default.conf
    restart: always

volumes:
  pgdata: