# ============================================================ # PDMP Unified Trading Platform — main.py # Version: v1.0.0 # Date: 2026-10-02 # ------------------------------------------------------------ # Unified backend serving: # 1. Sentiment Scanner (multi-source AI sentiment + FII/DII + # option chain + constituents + action dashboard) # 2. Zone Scanner (15-minute reversal zones on F&O stocks) # # Change log: # v1.0.0 - MERGE: Sentiment v0.8.1 + Zone Scanner v0.1.1 # Unified FastAPI app, single scheduler, dual DBs # v0.8.1 - Weight-aware + ATR-based tiered alerting (sentiment) # v0.1.1 - Sector tracking + corrected tickers (zone scanner) # v0.8.0 - Constituents heatmap (sentiment) # v0.7.x - FII/DII, option chain, OI buildup, YouTube, etc. # v0.1.0 - Zone scanner initial release # ============================================================ import os import re import time import math import sqlite3 import requests import subprocess import threading import json as json_lib import feedparser import urllib.parse from datetime import datetime, timedelta, timezone from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware import yfinance as yf from openai import OpenAI from youtube_transcript_api import YouTubeTranscriptApi from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger import pytz # Zone scanner imports from fno_stocks import get_unique_stocks as get_fno_stocks, MIN_PRICE, MAX_PRICE VERSION = "v1.0.0" # Cross-platform venv python if os.name == "nt": VENV_PYTHON = "python" else: VENV_PYTHON = "/home/dinkstrade/pdmp/venv/bin/python" IST = pytz.timezone("Asia/Kolkata") # ============================================================ # CONFIGURATION # ============================================================ NEWS_MAX_AGE_HOURS = 24 FII_DII_SCALE = 500 SUPPORT_RESISTANCE_SHIFT_MIN = 25 PCR_SHIFT_MIN = 0.05 VIX_SURGE_MIN_PCT = 3.0 ATR_SIGNAL_MULT = 0.25 ATR_IMPORTANT_MULT = 0.50 ATR_MAJOR_MULT = 0.85 TIER1_IMPORTANT_FLOOR = 0.5 TIER1_MAJOR_FLOOR = 0.9 ACTION_DASHBOARD_MAX_EVENTS = 25 # Zone scanner config PIVOT_LEFT = 2 PIVOT_RIGHT = 2 ZONE_LENGTH_BARS = 25 ZONE_MAX_AGE_HOURS = 72 # Extended from 24 to 72 to keep zones alive longer # Database paths SENTIMENT_DB_PATH = "sentiment_history.db" ZONES_DB_PATH = "zones_15m.db" # ============================================================ # NIFTY & SENSEX CONSTITUENTS # ============================================================ NIFTY50_CONSTITUENTS = [ {"symbol": "HDFCBANK.NS", "name": "HDFC Bank", "weight": 9.2, "sector": "Banks"}, {"symbol": "RELIANCE.NS", "name": "Reliance", "weight": 8.1, "sector": "Energy"}, {"symbol": "ICICIBANK.NS", "name": "ICICI Bank", "weight": 7.9, "sector": "Banks"}, {"symbol": "INFY.NS", "name": "Infosys", "weight": 5.8, "sector": "IT"}, {"symbol": "TCS.NS", "name": "TCS", "weight": 4.2, "sector": "IT"}, {"symbol": "ITC.NS", "name": "ITC", "weight": 3.8, "sector": "FMCG"}, {"symbol": "LT.NS", "name": "L&T", "weight": 3.5, "sector": "Infra"}, {"symbol": "SBIN.NS", "name": "SBI", "weight": 3.0, "sector": "Banks"}, {"symbol": "BHARTIARTL.NS", "name": "Bharti Airtel", "weight": 2.9, "sector": "Telecom"}, {"symbol": "AXISBANK.NS", "name": "Axis Bank", "weight": 2.7, "sector": "Banks"}, {"symbol": "KOTAKBANK.NS", "name": "Kotak Mahindra", "weight": 2.6, "sector": "Banks"}, {"symbol": "HINDUNILVR.NS", "name": "HUL", "weight": 2.5, "sector": "FMCG"}, {"symbol": "BAJFINANCE.NS", "name": "Bajaj Finance", "weight": 2.4, "sector": "NBFC"}, {"symbol": "MARUTI.NS", "name": "Maruti Suzuki", "weight": 2.2, "sector": "Auto"}, {"symbol": "ASIANPAINT.NS", "name": "Asian Paints", "weight": 2.0, "sector": "Paints"}, {"symbol": "M&M.NS", "name": "M&M", "weight": 1.8, "sector": "Auto"}, {"symbol": "SUNPHARMA.NS", "name": "Sun Pharma", "weight": 1.7, "sector": "Pharma"}, {"symbol": "HCLTECH.NS", "name": "HCL Tech", "weight": 1.6, "sector": "IT"}, {"symbol": "TITAN.NS", "name": "Titan", "weight": 1.5, "sector": "Consumer"}, {"symbol": "ULTRACEMCO.NS", "name": "UltraTech Cement", "weight": 1.4, "sector": "Cement"}, {"symbol": "WIPRO.NS", "name": "Wipro", "weight": 1.3, "sector": "IT"}, {"symbol": "ADANIENT.NS", "name": "Adani Enterprises", "weight": 1.3, "sector": "Conglomerate"}, {"symbol": "NTPC.NS", "name": "NTPC", "weight": 1.2, "sector": "Power"}, {"symbol": "BAJAJ-AUTO.NS", "name": "Bajaj Auto", "weight": 1.2, "sector": "Auto"}, {"symbol": "POWERGRID.NS", "name": "Power Grid", "weight": 1.2, "sector": "Power"}, ] SENSEX30_CONSTITUENTS = [ {"symbol": "HDFCBANK.NS", "name": "HDFC Bank", "weight": 13.5, "sector": "Banks"}, {"symbol": "RELIANCE.NS", "name": "Reliance", "weight": 12.0, "sector": "Energy"}, {"symbol": "ICICIBANK.NS", "name": "ICICI Bank", "weight": 9.0, "sector": "Banks"}, {"symbol": "INFY.NS", "name": "Infosys", "weight": 6.7, "sector": "IT"}, {"symbol": "TCS.NS", "name": "TCS", "weight": 5.2, "sector": "IT"}, {"symbol": "ITC.NS", "name": "ITC", "weight": 4.8, "sector": "FMCG"}, {"symbol": "LT.NS", "name": "L&T", "weight": 4.5, "sector": "Infra"}, {"symbol": "SBIN.NS", "name": "SBI", "weight": 3.7, "sector": "Banks"}, {"symbol": "BHARTIARTL.NS", "name": "Bharti Airtel", "weight": 3.5, "sector": "Telecom"}, {"symbol": "AXISBANK.NS", "name": "Axis Bank", "weight": 3.3, "sector": "Banks"}, {"symbol": "KOTAKBANK.NS", "name": "Kotak Mahindra", "weight": 3.2, "sector": "Banks"}, {"symbol": "HINDUNILVR.NS", "name": "HUL", "weight": 3.1, "sector": "FMCG"}, {"symbol": "BAJFINANCE.NS", "name": "Bajaj Finance", "weight": 2.9, "sector": "NBFC"}, {"symbol": "MARUTI.NS", "name": "Maruti Suzuki", "weight": 2.7, "sector": "Auto"}, {"symbol": "ASIANPAINT.NS", "name": "Asian Paints", "weight": 2.4, "sector": "Paints"}, {"symbol": "M&M.NS", "name": "M&M", "weight": 2.2, "sector": "Auto"}, {"symbol": "SUNPHARMA.NS", "name": "Sun Pharma", "weight": 2.1, "sector": "Pharma"}, {"symbol": "HCLTECH.NS", "name": "HCL Tech", "weight": 1.9, "sector": "IT"}, {"symbol": "TITAN.NS", "name": "Titan", "weight": 1.8, "sector": "Consumer"}, {"symbol": "ULTRACEMCO.NS", "name": "UltraTech Cement", "weight": 1.7, "sector": "Cement"}, {"symbol": "WIPRO.NS", "name": "Wipro", "weight": 1.6, "sector": "IT"}, {"symbol": "ADANIENT.NS", "name": "Adani Enterprises", "weight": 1.5, "sector": "Conglomerate"}, {"symbol": "NTPC.NS", "name": "NTPC", "weight": 1.4, "sector": "Power"}, {"symbol": "HEROMOTOCO.NS", "name": "Hero MotoCorp", "weight": 1.4, "sector": "Auto"}, {"symbol": "POWERGRID.NS", "name": "Power Grid", "weight": 1.4, "sector": "Power"}, {"symbol": "BAJAJFINSV.NS", "name": "Bajaj Finserv", "weight": 1.7, "sector": "NBFC"}, {"symbol": "INDUSINDBK.NS", "name": "IndusInd Bank", "weight": 1.3, "sector": "Banks"}, {"symbol": "NESTLEIND.NS", "name": "Nestle India", "weight": 1.2, "sector": "FMCG"}, {"symbol": "TECHM.NS", "name": "Tech Mahindra", "weight": 1.2, "sector": "IT"}, {"symbol": "TATASTEEL.NS", "name": "Tata Steel", "weight": 1.1, "sector": "Metals"}, ] TIER1_SYMBOLS = {"HDFCBANK.NS", "RELIANCE.NS", "ICICIBANK.NS"} SEARCH_QUERIES = { "NIFTY": '"Nifty 50" stock market India', "SENSEX": '"Sensex" OR "BSE Sensex" -defence -bankex -auto -PSU', "GOLD": '"gold rate" OR "gold price" India', "SILVER": '"silver rate" OR "silver price" India', "CRUDEOIL": '"Crude Oil" OR "Crude price" OR "Brent crude" India', "BITCOIN": '"Bitcoin price" cryptocurrency' } SYMBOLS = list(SEARCH_QUERIES.keys()) FII_DII_SYMBOLS = {"NIFTY", "SENSEX"} # ============================================================ # APP + CLIENT # ============================================================ app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) client = OpenAI( api_key=os.environ.get("GROQ_API_KEY"), base_url="https://api.groq.com/openai/v1" ) # ============================================================ # CACHE (shared by both scanners) # ============================================================ CACHE = {} def is_market_hours_now(): now = datetime.now(IST) if now.weekday() >= 5: return False market_open = now.replace(hour=9, minute=15, second=0, microsecond=0) market_close = now.replace(hour=15, minute=30, second=0, microsecond=0) return market_open <= now <= market_close def _auto_ttl_for_key(key): market_open = is_market_hours_now() if key.startswith("option_chain_"): return 180 if market_open else 1800 if key.startswith("constituents_"): return 180 if market_open else 1800 if key == "range": return 600 if market_open else 3600 if key == "fii_dii_score": return 900 if market_open else 3600 if key.startswith("sentiment_"): return 900 if market_open else 3600 if key == "latest_scan": return 300 if market_open else 1800 return 300 if market_open else 1800 def cache_get(key): if key in CACHE: value, timestamp, ttl = CACHE[key] if time.time() - timestamp < ttl: return value else: del CACHE[key] return None def cache_set(key, value, ttl_seconds=None): if ttl_seconds is None: ttl_seconds = _auto_ttl_for_key(key) CACHE[key] = (value, time.time(), ttl_seconds) def cache_peek(key): if key in CACHE: return CACHE[key][0] return None # ============================================================ # SENTIMENT DATABASE # ============================================================ def init_sentiment_db(): conn = sqlite3.connect(SENTIMENT_DB_PATH) cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS daily_sentiment ( date TEXT NOT NULL, symbol TEXT NOT NULL, score REAL NOT NULL, news_score REAL, twitter_score REAL, youtube_score REAL, fii_dii_score REAL, label TEXT, PRIMARY KEY (date, symbol) ) """) try: cursor.execute("ALTER TABLE daily_sentiment ADD COLUMN fii_dii_score REAL") except sqlite3.OperationalError: pass conn.commit() conn.close() def save_to_history(symbol: str, result: dict): today = datetime.now(IST).strftime("%Y-%m-%d") try: conn = sqlite3.connect(SENTIMENT_DB_PATH) cursor = conn.cursor() cursor.execute(""" INSERT OR REPLACE INTO daily_sentiment (date, symbol, score, news_score, twitter_score, youtube_score, fii_dii_score, label) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( today, symbol, result.get("score", 0), result.get("news_score", 0), result.get("twitter_score", 0), result.get("youtube_score", 0), result.get("fii_dii_score", 0), result.get("label", "NEUTRAL") )) conn.commit() conn.close() except Exception as e: print(f"Sentiment DB save error for {symbol}: {e}") def get_history(symbol: str, days: int = 7) -> list: try: conn = sqlite3.connect(SENTIMENT_DB_PATH) cursor = conn.cursor() cursor.execute(""" SELECT date, score, label FROM daily_sentiment WHERE symbol = ? ORDER BY date DESC LIMIT ? """, (symbol, days)) rows = cursor.fetchall() conn.close() return [{"date": r[0], "score": r[1], "label": r[2]} for r in reversed(rows)] except Exception as e: print(f"Sentiment DB read error for {symbol}: {e}") return [] init_sentiment_db() # ============================================================ # SENTIMENT — fallback keyword scorer # ============================================================ BEARISH_WORDS = ['fall', 'falls', 'crash', 'crashes', 'plunge', 'plunges', 'drop', 'drops', 'decline', 'declines', 'slump', 'slumps', 'weak', 'loss', 'losses', 'down', 'bearish', 'sell-off', 'selloff', 'tumble', 'tumbles', 'slide'] BULLISH_WORDS = ['rise', 'rises', 'surge', 'surges', 'gain', 'gains', 'jump', 'jumps', 'rally', 'rallies', 'high', 'record', 'strong', 'up', 'bullish', 'soar', 'soars', 'climb', 'climbs', 'recover', 'recovers'] def keyword_score(text: str) -> float: text_lower = text.lower() bull = sum(text_lower.count(w) for w in BULLISH_WORDS) bear = sum(text_lower.count(w) for w in BEARISH_WORDS) total = bull + bear if total == 0: return 0.0 score = ((bull - bear) / total) * 10 return round(max(-10, min(10, score)), 1) # ============================================================ # SENTIMENT — AI engine # ============================================================ def ai_sentiment(symbol: str, text_block: str) -> float: if not os.environ.get("GROQ_API_KEY") or not text_block.strip(): return keyword_score(text_block) prompt = f"""Analyze the sentiment of the following text about {symbol}. Rate from -10 (very bearish) to +10 (very bullish). Reply with only a number. Text: {text_block[:3000]} Sentiment score:""" try: response = client.chat.completions.create( model="openai/gpt-oss-120b", messages=[ {"role": "user", "content": "Analyze: 'Nifty hits all-time high on strong earnings'"}, {"role": "assistant", "content": "7"}, {"role": "user", "content": "Analyze: 'Market crashes on weak global cues'"}, {"role": "assistant", "content": "-6"}, {"role": "user", "content": prompt} ], temperature=0.3, max_tokens=50, extra_body={"reasoning_effort": "low"} ) raw = response.choices[0].message.content.strip() match = re.search(r'-?\d+\.?\d*', raw) if match: return round(max(-10.0, min(10.0, float(match.group()))), 1) else: return keyword_score(text_block) except Exception as e: print(f"[{symbol}] AI error: {e}. Using keyword fallback.") return keyword_score(text_block) # ============================================================ # SENTIMENT — freshness filter # ============================================================ def is_fresh(entry, max_age_hours: int = NEWS_MAX_AGE_HOURS) -> bool: published = getattr(entry, "published_parsed", None) if not published: return True try: published_dt = datetime(*published[:6], tzinfo=timezone.utc) except Exception: return True cutoff = datetime.now(timezone.utc) - timedelta(hours=max_age_hours) return published_dt >= cutoff # ============================================================ # SENTIMENT — FII/DII # ============================================================ def fetch_fii_dii_score() -> float: cache_key = "fii_dii_score" cached = cache_get(cache_key) if cached is not None: return cached try: headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Accept": "application/json", "Referer": "https://www.nseindia.com/" } r = requests.get( "https://www.nseindia.com/api/fiidiiTradeReact", headers=headers, timeout=10 ) if r.status_code != 200: print(f"⚠️ FII/DII NSE returned status {r.status_code}") cache_set(cache_key, 0.0) return 0.0 data = r.json() if not isinstance(data, list) or len(data) == 0: cache_set(cache_key, 0.0) return 0.0 fii_net = None dii_net = None for row in data: category = str(row.get("category", "")).upper() net_str = str(row.get("netValue", "0")).replace(",", "").strip() try: net = float(net_str) except ValueError: net = 0.0 if "FII" in category or "FPI" in category: fii_net = net elif "DII" in category: dii_net = net if fii_net is None or dii_net is None: cache_set(cache_key, 0.0) return 0.0 combined_net = fii_net + dii_net score = max(-10.0, min(10.0, combined_net / FII_DII_SCALE)) score = round(score, 1) print(f"[FII/DII] FII net: {fii_net:+.2f} cr | DII net: {dii_net:+.2f} cr | " f"Combined: {combined_net:+.2f} cr | Score: {score}") cache_set(cache_key, score) return score except Exception as e: print(f"⚠️ FII/DII fetch error: {e}") cache_set(cache_key, 0.0) return 0.0 # ============================================================ # SENTIMENT — option chain # ============================================================ def fetch_option_chain_levels(symbol: str = "NIFTY") -> dict: cache_key = f"option_chain_{symbol}" cached = cache_get(cache_key) if cached is not None: return cached try: helper_path = os.path.join( os.path.dirname(os.path.abspath(__file__)), "option_chain_helper.py" ) result = subprocess.run( [VENV_PYTHON, helper_path, symbol], capture_output=True, text=True, timeout=30 ) if result.returncode != 0: err = result.stderr.strip() or "unknown error" print(f"⚠️ Option chain helper failed: {err[:200]}") cache_set(cache_key, {"error": err[:200]}) return {"error": err[:200]} data = json_lib.loads(result.stdout.strip()) cache_set(cache_key, data) print(f"[Option Chain] {symbol} spot={data.get('spot')} " f"R={data.get('resistance',{}).get('level')} " f"S={data.get('support',{}).get('level')} " f"PCR={data.get('pcr')}") return data except subprocess.TimeoutExpired: print(f"⚠️ Option chain helper timed out for {symbol}") cache_set(cache_key, {"error": "timeout"}) return {"error": "timeout"} except Exception as e: print(f"⚠️ Option chain error: {e}") cache_set(cache_key, {"error": str(e)}) return {"error": str(e)} # ============================================================ # SENTIMENT — safe helpers # ============================================================ def _safe_float(val, decimals=2): try: if val is None: return None f = float(val) if math.isnan(f) or math.isinf(f): return None return round(f, decimals) except (ValueError, TypeError): return None def _safe_int(val): try: if val is None: return None f = float(val) if math.isnan(f) or math.isinf(f): return None return int(f) except (ValueError, TypeError): return None # ============================================================ # SENTIMENT — constituents # ============================================================ def fetch_constituents(index: str = "NIFTY") -> dict: cache_key = f"constituents_{index}" cached = cache_get(cache_key) if cached is not None: return cached try: constituents = NIFTY50_CONSTITUENTS if index == "NIFTY" else SENSEX30_CONSTITUENTS symbols = [c["symbol"] for c in constituents] data = yf.download( tickers=" ".join(symbols), period="7d", interval="1d", progress=False, group_by="ticker", auto_adjust=False, threads=True ) results = [] for c in constituents: symbol = c["symbol"] try: if len(symbols) == 1: ticker_data = data else: ticker_data = data[symbol] if symbol in data.columns.get_level_values(0) else None if ticker_data is None or ticker_data.empty: results.append({ "symbol": symbol, "name": c["name"], "weight": c["weight"], "sector": c["sector"], "price": None, "change_pct": None, "atr": None, "atr_pct": None, "move_vs_atr": None, "volume": None, "day_high": None, "day_low": None, "error": "no_data" }) continue closes = ticker_data["Close"].dropna() highs = ticker_data["High"].dropna() if "High" in ticker_data else None lows = ticker_data["Low"].dropna() if "Low" in ticker_data else None if len(closes) < 2: results.append({ "symbol": symbol, "name": c["name"], "weight": c["weight"], "sector": c["sector"], "price": _safe_float(closes.iloc[-1]) if len(closes) else None, "change_pct": None, "atr": None, "atr_pct": None, "move_vs_atr": None, "volume": None, "day_high": None, "day_low": None, "error": "insufficient_history" }) continue prev_close = float(closes.iloc[-2]) current_price = float(closes.iloc[-1]) change_pct = round(((current_price - prev_close) / prev_close) * 100, 2) atr = None atr_pct = None move_vs_atr = None if highs is not None and lows is not None and len(highs) >= 5 and len(lows) >= 5: true_ranges = [] for i in range(-5, 0): try: h = float(highs.iloc[i]) l = float(lows.iloc[i]) tr = h - l if tr > 0: true_ranges.append(tr) except Exception: continue if true_ranges: atr = round(sum(true_ranges) / len(true_ranges), 2) if current_price > 0: atr_pct = round((atr / current_price) * 100, 2) if atr_pct > 0: move_vs_atr = round(abs(change_pct) / atr_pct, 2) day_high = _safe_float(highs.iloc[-1]) if highs is not None and not highs.empty else None day_low = _safe_float(lows.iloc[-1]) if lows is not None and not lows.empty else None volume = _safe_int(ticker_data["Volume"].iloc[-1]) if "Volume" in ticker_data and not ticker_data["Volume"].empty else None results.append({ "symbol": symbol, "name": c["name"], "weight": c["weight"], "sector": c["sector"], "price": round(current_price, 2), "change_pct": change_pct, "prev_close": round(prev_close, 2), "atr": atr, "atr_pct": atr_pct, "move_vs_atr": move_vs_atr, "day_high": day_high, "day_low": day_low, "volume": volume, "error": None }) except Exception as ex: results.append({ "symbol": symbol, "name": c["name"], "weight": c["weight"], "sector": c["sector"], "price": None, "change_pct": None, "atr": None, "atr_pct": None, "move_vs_atr": None, "volume": None, "day_high": None, "day_low": None, "error": str(ex)[:80] }) result = { "index": index, "updated": datetime.now(IST).strftime("%d %b %Y, %I:%M %p IST"), "constituents": results, "count": len(results) } cache_set(cache_key, result) print(f"[Constituents] {index}: fetched {len(results)} stocks") return result except Exception as e: print(f"⚠️ Constituents fetch error for {index}: {e}") err = {"index": index, "error": str(e)[:200], "constituents": []} cache_set(cache_key, err) return err # ============================================================ # SENTIMENT — data sources # ============================================================ def fetch_news_headlines(symbol: str, max_headlines: int = 7) -> list: query = urllib.parse.quote(SEARCH_QUERIES.get(symbol, symbol)) url = f"https://news.google.com/rss/search?q={query}&hl=en-IN&gl=IN&ceid=IN:en" feed = feedparser.parse(url) fresh = [e for e in feed.entries if is_fresh(e)] return [entry.title for entry in fresh[:max_headlines]] def fetch_twitter_posts(symbol: str, max_posts: int = 5) -> list: keyword = SEARCH_QUERIES.get(symbol, symbol) query = urllib.parse.quote(f'{keyword} site:x.com OR site:twitter.com') url = f"https://news.google.com/rss/search?q={query}&hl=en-IN&gl=IN&ceid=IN:en" feed = feedparser.parse(url) fresh = [e for e in feed.entries if is_fresh(e)] return [entry.title for entry in fresh[:max_posts]] # ============================================================ # SENTIMENT — YouTube # ============================================================ FINANCE_CHANNELS = {"CNBC-TV18": "cnbctv18", "ET Now": "ETNOW"} SYMBOL_KEYWORDS = { "NIFTY": ["nifty", "nifty50", "nifty 50"], "SENSEX": ["sensex", "bse sensex"], "GOLD": ["gold", "gold price", "gold rate", "mcx gold"], "SILVER": ["silver", "silver price", "silver rate"], "CRUDEOIL": ["crude", "crude oil", "brent", "wti"], "BITCOIN": ["bitcoin", "btc"], } def extract_video_id(url: str) -> str: match = re.search(r'(?:v=|youtu\.be/|/embed/|/shorts/)([A-Za-z0-9_-]{11})', url) return match.group(1) if match else "" def fetch_youtube_transcript(symbol: str, max_chars: int = 2500) -> str: keywords = SYMBOL_KEYWORDS.get(symbol, [symbol.lower()]) cutoff = datetime.now(timezone.utc) - timedelta(hours=NEWS_MAX_AGE_HOURS) try: api = YouTubeTranscriptApi() except Exception as e: print(f"[{symbol}] YouTube init error: {e}") return "" for channel_name, username in FINANCE_CHANNELS.items(): try: feed_url = f"https://www.youtube.com/feeds/videos.xml?user={username}" feed = feedparser.parse(feed_url) for entry in feed.entries: published = getattr(entry, "published_parsed", None) if not published: continue published_dt = datetime(*published[:6], tzinfo=timezone.utc) if published_dt < cutoff: continue title_lower = entry.title.lower() if not any(kw in title_lower for kw in keywords): continue video_id = extract_video_id(entry.link) if not video_id: continue try: transcript = api.fetch(video_id, languages=['en', 'en-IN', 'hi']) parts = [] for chunk in transcript: text = getattr(chunk, 'text', None) if text: parts.append(text) full_text = ' '.join(parts) print(f"[{symbol}] YouTube OK '{entry.title[:60]}' from {channel_name}") return full_text[:max_chars] except Exception: continue except Exception as e: print(f"[{symbol}] Channel {channel_name} error: {e}") continue print(f"[{symbol}] YouTube: no usable transcript found") return "" # ============================================================ # SENTIMENT — main aggregator # ============================================================ def get_sentiment_score(symbol: str, bypass_cache: bool = False) -> dict: cache_key = f"sentiment_{symbol}" if not bypass_cache: cached = cache_get(cache_key) if cached: return cached news = fetch_news_headlines(symbol) tweets = fetch_twitter_posts(symbol) youtube_text = fetch_youtube_transcript(symbol) fii_dii_score = fetch_fii_dii_score() if symbol in FII_DII_SYMBOLS else 0.0 news_score = ai_sentiment(symbol, '\n'.join(news)) if news else 0.0 twitter_score = ai_sentiment(symbol, '\n'.join(tweets)) if tweets else 0.0 youtube_score = ai_sentiment(symbol, youtube_text) if youtube_text else 0.0 weights, scores = [], [] if news: weights.append(0.50); scores.append(news_score) if youtube_text: weights.append(0.20); scores.append(youtube_score) if tweets: weights.append(0.10); scores.append(twitter_score) if symbol in FII_DII_SYMBOLS: weights.append(0.30); scores.append(fii_dii_score) if not scores: result = { "symbol": symbol, "score": 0.0, "label": "NEUTRAL", "news_score": 0.0, "twitter_score": 0.0, "youtube_score": 0.0, "fii_dii_score": 0.0, "headlines": [], "tweets": [], "has_youtube": False } cache_set(cache_key, result) save_to_history(symbol, result) return result total_weight = sum(weights) final_score = round(sum(s * w for s, w in zip(scores, weights)) / total_weight, 1) if final_score <= -1: label = "BEARISH" elif final_score >= 1: label = "BULLISH" else: label = "NEUTRAL" result = { "symbol": symbol, "score": final_score, "label": label, "news_score": news_score, "twitter_score": twitter_score, "youtube_score": youtube_score, "fii_dii_score": fii_dii_score, "headlines": news, "tweets": tweets, "has_youtube": bool(youtube_text) } cache_set(cache_key, result) save_to_history(symbol, result) return result # ============================================================ # ACTION DASHBOARD (sentiment) # ============================================================ PREVIOUS_SNAPSHOT = {} def _build_sector_summary(constituents: list) -> dict: sector_data = {} for c in constituents: if c.get("change_pct") is None: continue s = c["sector"] if s not in sector_data: sector_data[s] = {"count": 0, "avg_change": 0.0, "total_weight": 0.0, "stocks": []} sector_data[s]["count"] += 1 sector_data[s]["avg_change"] += c["change_pct"] sector_data[s]["total_weight"] += c["weight"] sector_data[s]["stocks"].append(c["name"]) for s in sector_data: if sector_data[s]["count"] > 0: sector_data[s]["avg_change"] = round(sector_data[s]["avg_change"] / sector_data[s]["count"], 2) return sector_data def _classify_move(stock: dict) -> dict: pct = stock.get("change_pct") atr_pct = stock.get("atr_pct") symbol = stock.get("symbol", "") is_tier1 = symbol in TIER1_SYMBOLS if pct is None: return {"level": None, "suggested_action": None} abs_pct = abs(pct) if is_tier1: if abs_pct >= TIER1_MAJOR_FLOOR: return {"level": "MAJOR", "suggested_action": "Extreme move for a heavyweight — momentum may exhaust soon, use caution"} if abs_pct >= TIER1_IMPORTANT_FLOOR: return {"level": "IMPORTANT", "suggested_action": "Significant heavyweight move — chart may show entry opportunity"} if atr_pct is None or atr_pct <= 0: if abs_pct >= 5.0: return {"level": "MAJOR", "suggested_action": "Extreme move — use caution"} if abs_pct >= 3.0: return {"level": "IMPORTANT", "suggested_action": "Significant move — watch for continuation"} if abs_pct >= 1.5: return {"level": "SIGNAL", "suggested_action": "Early mover — open chart, watch for continuation"} return {"level": None, "suggested_action": None} move_vs_atr = abs_pct / atr_pct if move_vs_atr >= ATR_MAJOR_MULT: return {"level": "MAJOR", "suggested_action": "Extreme move — momentum may exhaust soon, use caution"} if move_vs_atr >= ATR_IMPORTANT_MULT: return {"level": "IMPORTANT", "suggested_action": "Significant move — chart may show entry opportunity"} if move_vs_atr >= ATR_SIGNAL_MULT: return {"level": "SIGNAL", "suggested_action": "Early mover — open chart, watch for continuation"} return {"level": None, "suggested_action": None} def generate_action_events() -> list: events = [] curr_sup = None; curr_res = None; curr_pcr = None; curr_maxpain = None current_vix = None; vix_change_pct = 0.0 nifty_oc = fetch_option_chain_levels("NIFTY") try: vix_ticker = yf.Ticker("^INDIAVIX") vix_data = vix_ticker.history(period="5d") if not vix_data.empty: current_vix = float(vix_data["Close"].iloc[-1]) if len(vix_data) >= 2: prev_vix = float(vix_data["Close"].iloc[-2]) if prev_vix: vix_change_pct = round(((current_vix - prev_vix) / prev_vix) * 100, 2) except Exception as e: print(f"[Action] VIX error: {e}") nifty_const = fetch_constituents("NIFTY").get("constituents", []) if current_vix is not None and abs(vix_change_pct) >= VIX_SURGE_MIN_PCT: if vix_change_pct > 0: events.append({"priority": 1, "icon": "⚠️", "type": "vix_surge", "level": "MAJOR", "message": f"VIX surged +{vix_change_pct:.1f}% to {current_vix:.2f} — premium expansion warning", "suggested_action": "Options getting expensive — avoid chasing premium"}) else: events.append({"priority": 1, "icon": "✅", "type": "vix_drop", "level": "IMPORTANT", "message": f"VIX dropped {vix_change_pct:.1f}% to {current_vix:.2f} — premium contraction", "suggested_action": "Calmer market — options cheaper, potential entry"}) for stock in nifty_const: pct = stock.get("change_pct") if pct is None: continue weight = stock.get("weight", 0) impact = round(abs(pct) * weight / 100, 3) move_vs_atr = stock.get("move_vs_atr") direction = "up" if pct > 0 else "down" classification = _classify_move(stock) level = classification.get("level") suggested_action = classification.get("suggested_action") if level == "MAJOR": icon, priority = "🚨", 1 elif level == "IMPORTANT": icon, priority = "🔔", 2 elif level == "SIGNAL": icon, priority = "🔍", 3 else: continue atr_note = f" • {int(move_vs_atr * 100)}% of 5-day ATR" if move_vs_atr is not None else "" impact_note = f" • contributing ~{impact:.3f}% to NIFTY" message = f"{level}: {stock['name']} {pct:+.2f}% ({weight}% weight) — {direction}{atr_note}{impact_note}" events.append({"priority": priority, "icon": icon, "type": f"{level.lower()}_move", "level": level, "message": message, "suggested_action": suggested_action}) sector_summary = _build_sector_summary(nifty_const) for sector, sdata in sector_summary.items(): if sdata["count"] >= 3 and abs(sdata["avg_change"]) >= 1.5: direction = "weak" if sdata["avg_change"] < 0 else "strong" events.append({"priority": 4, "icon": "📊", "type": "sector_alert", "level": "IMPORTANT", "message": f"{sector} sector {direction}: avg {sdata['avg_change']:+.2f}% across {sdata['count']} stocks", "suggested_action": f"Coordinated {sector} move — check sector rotation"}) if "support" in nifty_oc and isinstance(nifty_oc.get("support"), dict): curr_sup = nifty_oc["support"].get("level") prev_sup = PREVIOUS_SNAPSHOT.get("nifty_support") if curr_sup is not None and prev_sup is not None: diff = curr_sup - prev_sup if abs(diff) >= SUPPORT_RESISTANCE_SHIFT_MIN: direction = "UP" if diff > 0 else "DOWN" events.append({"priority": 4, "icon": "📍", "type": "support_shift", "level": "IMPORTANT", "message": f"NIFTY Support shifted {direction} from {int(prev_sup):,} to {int(curr_sup):,}", "suggested_action": "Adjust stops and targets to new support"}) if "resistance" in nifty_oc and isinstance(nifty_oc.get("resistance"), dict): curr_res = nifty_oc["resistance"].get("level") prev_res = PREVIOUS_SNAPSHOT.get("nifty_resistance") if curr_res is not None and prev_res is not None: diff = curr_res - prev_res if abs(diff) >= SUPPORT_RESISTANCE_SHIFT_MIN: direction = "UP" if diff > 0 else "DOWN" events.append({"priority": 4, "icon": "📍", "type": "resistance_shift", "level": "IMPORTANT", "message": f"NIFTY Resistance shifted {direction} from {int(prev_res):,} to {int(curr_res):,}", "suggested_action": "Adjust stops and targets to new resistance"}) curr_pcr = nifty_oc.get("pcr") prev_pcr = PREVIOUS_SNAPSHOT.get("nifty_pcr") if curr_pcr is not None and prev_pcr is not None: diff = curr_pcr - prev_pcr if abs(diff) >= PCR_SHIFT_MIN: tone = "bullish" if diff > 0 else "bearish" events.append({"priority": 5, "icon": "🔄", "type": "pcr_shift", "level": "SIGNAL", "message": f"PCR moved from {prev_pcr:.2f} to {curr_pcr:.2f} — {tone} shift", "suggested_action": f"Positioning is now {tone}"}) crude_sent = cache_peek("sentiment_CRUDEOIL") if crude_sent and crude_sent.get("score") is not None and crude_sent["score"] <= -3: events.append({"priority": 6, "icon": "🛢️", "type": "crude_warning", "level": "SIGNAL", "message": f"Crudeoil bearish (score {crude_sent['score']}) — Reliance may follow", "suggested_action": "Watch Reliance for sympathetic move"}) gold_sent = cache_peek("sentiment_GOLD") if gold_sent and gold_sent.get("score") is not None and gold_sent["score"] >= 3: events.append({"priority": 6, "icon": "🥇", "type": "gold_strength", "level": "SIGNAL", "message": f"Gold bullish (score {gold_sent['score']}) — defensive rotation", "suggested_action": "Watch for risk-off trade developing"}) curr_maxpain = nifty_oc.get("max_pain") prev_maxpain = PREVIOUS_SNAPSHOT.get("nifty_maxpain") if curr_maxpain is not None and prev_maxpain is not None: diff = curr_maxpain - prev_maxpain if abs(diff) >= 50: events.append({"priority": 7, "icon": "🎯", "type": "maxpain_shift", "level": "SIGNAL", "message": f"Max Pain shifted from {int(prev_maxpain):,} to {int(curr_maxpain):,}", "suggested_action": "Expiry target has moved"}) if curr_sup is not None: PREVIOUS_SNAPSHOT["nifty_support"] = curr_sup if curr_res is not None: PREVIOUS_SNAPSHOT["nifty_resistance"] = curr_res if curr_pcr is not None: PREVIOUS_SNAPSHOT["nifty_pcr"] = curr_pcr if curr_maxpain is not None: PREVIOUS_SNAPSHOT["nifty_maxpain"] = curr_maxpain events.sort(key=lambda x: x["priority"]) events = events[:ACTION_DASHBOARD_MAX_EVENTS] if not events: events.append({"priority": 99, "icon": "⚪", "type": "neutral", "level": "INFO", "message": "No significant events detected — market is quiet", "suggested_action": None}) return events # ============================================================ # ZONE SCANNER — Database # ============================================================ def init_zones_db(): conn = sqlite3.connect(ZONES_DB_PATH) c = conn.cursor() c.execute(""" CREATE TABLE IF NOT EXISTS zones ( id INTEGER PRIMARY KEY AUTOINCREMENT, symbol TEXT NOT NULL, name TEXT, sector TEXT, zone_type TEXT NOT NULL, zone_top REAL NOT NULL, zone_bottom REAL NOT NULL, created_at TEXT NOT NULL, created_price REAL, age_bars INTEGER DEFAULT 0, entry_count INTEGER DEFAULT 0, first_entry_at TEXT, last_entry_at TEXT, last_entry_price REAL, is_active INTEGER DEFAULT 1, UNIQUE(symbol, zone_type, created_at) ) """) c.execute(""" CREATE TABLE IF NOT EXISTS zone_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, symbol TEXT NOT NULL, name TEXT, sector TEXT, event_type TEXT NOT NULL, zone_type TEXT, zone_top REAL, zone_bottom REAL, price REAL, event_at TEXT NOT NULL, tier TEXT, message TEXT ) """) try: c.execute("ALTER TABLE zone_events ADD COLUMN sector TEXT") except sqlite3.OperationalError: pass conn.commit() conn.close() init_zones_db() def zone_exists(symbol, zone_type, created_at): conn = sqlite3.connect(ZONES_DB_PATH) c = conn.cursor() c.execute("SELECT id FROM zones WHERE symbol=? AND zone_type=? AND created_at=?", (symbol, zone_type, created_at)) row = c.fetchone() conn.close() return row is not None def insert_zone(zone): conn = sqlite3.connect(ZONES_DB_PATH) c = conn.cursor() try: c.execute(""" INSERT OR IGNORE INTO zones (symbol, name, sector, zone_type, zone_top, zone_bottom, created_at, created_price, age_bars, is_active) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, 1) """, ( zone["symbol"], zone["name"], zone.get("sector", ""), zone["zone_type"], zone["zone_top"], zone["zone_bottom"], zone["created_at"], zone["created_price"] )) conn.commit() except Exception as e: print(f"Insert zone error: {e}") conn.close() def insert_zone_event(event): conn = sqlite3.connect(ZONES_DB_PATH) c = conn.cursor() try: c.execute(""" INSERT INTO zone_events (symbol, name, sector, event_type, zone_type, zone_top, zone_bottom, price, event_at, tier, message) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( event["symbol"], event.get("name"), event.get("sector", ""), event["event_type"], event.get("zone_type"), event.get("zone_top"), event.get("zone_bottom"), event.get("price"), event["event_at"], event.get("tier"), event.get("message") )) conn.commit() except Exception as e: print(f"Insert zone event error: {e}") conn.close() def get_active_zones(): cutoff = (datetime.now(IST) - timedelta(hours=ZONE_MAX_AGE_HOURS)).isoformat() conn = sqlite3.connect(ZONES_DB_PATH) c = conn.cursor() c.execute(""" SELECT id, symbol, name, sector, zone_type, zone_top, zone_bottom, created_at, created_price, age_bars, entry_count, first_entry_at, last_entry_at, last_entry_price FROM zones WHERE is_active = 1 AND created_at >= ? """, (cutoff,)) rows = c.fetchall() conn.close() return [{ "id": r[0], "symbol": r[1], "name": r[2], "sector": r[3], "zone_type": r[4], "zone_top": r[5], "zone_bottom": r[6], "created_at": r[7], "created_price": r[8], "age_bars": r[9], "entry_count": r[10], "first_entry_at": r[11], "last_entry_at": r[12], "last_entry_price": r[13], } for r in rows] def update_zone_entry(zone_id, price, at_time): conn = sqlite3.connect(ZONES_DB_PATH) c = conn.cursor() c.execute(""" UPDATE zones SET entry_count = entry_count + 1, first_entry_at = COALESCE(first_entry_at, ?), last_entry_at = ?, last_entry_price = ? WHERE id = ? """, (at_time, at_time, price, zone_id)) conn.commit() conn.close() def deactivate_old_zones(): cutoff = (datetime.now(IST) - timedelta(hours=ZONE_MAX_AGE_HOURS)).isoformat() conn = sqlite3.connect(ZONES_DB_PATH) c = conn.cursor() c.execute("UPDATE zones SET is_active = 0 WHERE created_at < ?", (cutoff,)) conn.commit() conn.close() # ============================================================ # ZONE SCANNER — pivot detection # ============================================================ def find_pivots(bars): pivots = [] n = len(bars) if n < PIVOT_LEFT + PIVOT_RIGHT + 1: return pivots for i in range(PIVOT_LEFT, n - PIVOT_RIGHT): bar = bars[i] is_pivot_low = True for j in range(i - PIVOT_LEFT, i): if bars[j]["low"] <= bar["low"]: is_pivot_low = False break if is_pivot_low: for j in range(i + 1, i + PIVOT_RIGHT + 1): if bars[j]["low"] <= bar["low"]: is_pivot_low = False break if is_pivot_low: body_low = min(bar["open"], bar["close"]) box_top = max(bar["low"], body_low) box_bottom = min(bar["low"], body_low) pivots.append({"index": i, "type": "bull", "top": round(box_top, 2), "bottom": round(box_bottom, 2), "price": bar["low"], "time": bar["time"]}) is_pivot_high = True for j in range(i - PIVOT_LEFT, i): if bars[j]["high"] >= bar["high"]: is_pivot_high = False break if is_pivot_high: for j in range(i + 1, i + PIVOT_RIGHT + 1): if bars[j]["high"] >= bar["high"]: is_pivot_high = False break if is_pivot_high: body_high = max(bar["open"], bar["close"]) box_top = max(bar["high"], body_high) box_bottom = min(bar["high"], body_high) pivots.append({"index": i, "type": "bear", "top": round(box_top, 2), "bottom": round(box_bottom, 2), "price": bar["high"], "time": bar["time"]}) return pivots # ============================================================ # ZONE SCANNER — data fetch # ============================================================ def fetch_15m_bars(symbol, days=3): try: ticker = yf.Ticker(symbol) df = ticker.history(period=f"{days}d", interval="15m", prepost=False) if df.empty: return [] bars = [] for idx, row in df.iterrows(): try: bars.append({ "time": idx.isoformat(), "open": float(row["Open"]), "high": float(row["High"]), "low": float(row["Low"]), "close": float(row["Close"]), "volume": int(row["Volume"]) if not math.isnan(float(row["Volume"])) else 0, }) except (ValueError, TypeError): continue return bars except Exception as e: print(f"[{symbol}] 15m fetch error: {e}") return [] # ============================================================ # ZONE SCANNER — tier classifier # ============================================================ def classify_zone_event(stock, zone, event_type, is_fresh=True): name = stock.get("name", stock["symbol"]) zone_type = zone.get("zone_type") zone_label = "🟢 Bullish" if zone_type == "bull" else "🔴 Bearish" if event_type == "zone_created": tier = "IMPORTANT" if is_fresh else "SIGNAL" msg = f"NEW ZONE: {name} created a {zone_label} reversal zone ({zone['zone_bottom']}-{zone['zone_top']})" elif event_type == "zone_entered": entry_count = zone.get("entry_count", 0) if entry_count == 0: tier = "IMPORTANT" msg = f"FIRST ENTRY: {name} entered {zone_label} reversal zone ({zone['zone_bottom']}-{zone['zone_top']})" else: tier = "SIGNAL" msg = f"RE-ENTRY: {name} re-entered {zone_label} zone (entry #{entry_count + 1})" else: tier = "SIGNAL" msg = f"{name} zone event" return tier, msg # ============================================================ # ZONE SCANNER — main scan # ============================================================ def run_zone_scan(min_price=None, max_price=None): started = datetime.now(IST) _min = min_price if min_price is not None else MIN_PRICE _max = max_price if max_price is not None else MAX_PRICE print(f"\n[{started.strftime('%Y-%m-%d %H:%M:%S IST')}] Zone scan started " f"(price filter: ₹{_min}-₹{_max})...") stocks = get_fno_stocks() print(f" Universe: {len(stocks)} F&O stocks") deactivate_old_zones() new_events = [] active_zones_by_symbol = {} for z in get_active_zones(): active_zones_by_symbol.setdefault(z["symbol"], []).append(z) scanned_count = 0 pivots_found = 0 entries_found = 0 for stock in stocks: symbol = stock["symbol"] try: bars = fetch_15m_bars(symbol, days=3) if len(bars) < 10: continue current_price = bars[-1]["close"] if current_price < _min or current_price > _max: continue scanned_count += 1 pivots = find_pivots(bars) if not pivots: continue fresh_pivots = [p for p in pivots if p["index"] >= len(bars) - PIVOT_RIGHT - 3] for p in fresh_pivots: created_at = p["time"] if zone_exists(symbol, p["type"], created_at): continue pivots_found += 1 zone = { "symbol": symbol, "name": stock["name"], "sector": stock.get("sector", ""), "zone_type": p["type"], "zone_top": p["top"], "zone_bottom": p["bottom"], "created_at": created_at, "created_price": p["price"], } insert_zone(zone) tier, msg = classify_zone_event(stock, zone, "zone_created", is_fresh=True) new_events.append({ "symbol": symbol, "name": stock["name"], "sector": stock.get("sector", ""), "event_type": "zone_created", "zone_type": p["type"], "zone_top": p["top"], "zone_bottom": p["bottom"], "price": current_price, "event_at": started.isoformat(), "tier": tier, "message": msg, }) symbol_zones = active_zones_by_symbol.get(symbol, []) for zone in symbol_zones: zt = zone["zone_top"]; zb = zone["zone_bottom"] for bar in bars[-3:]: touched = ( (bar["low"] <= zt and bar["low"] >= zb) or (bar["high"] >= zb and bar["high"] <= zt) or (bar["low"] <= zb and bar["high"] >= zt) ) if touched: update_zone_entry(zone["id"], bar["close"], bar["time"]) tier, msg = classify_zone_event(stock, zone, "zone_entered") new_events.append({ "symbol": symbol, "name": stock["name"], "sector": stock.get("sector", ""), "event_type": "zone_entered", "zone_type": zone["zone_type"], "zone_top": zt, "zone_bottom": zb, "price": bar["close"], "event_at": bar["time"], "tier": tier, "message": msg, }) entries_found += 1 break except Exception as e: print(f"[{symbol}] scan error: {e}") continue for evt in new_events: insert_zone_event(evt) elapsed = (datetime.now(IST) - started).total_seconds() print(f" Scanned: {scanned_count}/{len(stocks)} stocks") print(f" New zones: {pivots_found} | Entries: {entries_found}") print(f" Completed in {elapsed:.1f}s") result = { "version": VERSION, "scanned_at": started.isoformat(), "scanned_at_human": started.strftime("%d %b %Y, %I:%M:%S %p IST"), "stocks_scanned": scanned_count, "new_zones_count": len([e for e in new_events if e["event_type"] == "zone_created"]), "entries_count": len([e for e in new_events if e["event_type"] == "zone_entered"]), "new_zones": [e for e in new_events if e["event_type"] == "zone_created"], "recent_entries": [e for e in new_events if e["event_type"] == "zone_entered"], "active_zones_total": len(get_active_zones()), } cache_set("latest_scan", result) return result # ============================================================ # SCHEDULED REFRESH (sentiment) # ============================================================ def scheduled_sentiment_refresh(): print(f"\n[{datetime.now(IST).strftime('%Y-%m-%d %H:%M:%S IST')}] Scheduled sentiment refresh...") for sym in SYMBOLS: try: result = get_sentiment_score(sym, bypass_cache=True) print(f" [{sym}] Score: {result['score']} ({result['label']})") except Exception as e: print(f" [{sym}] error: {e}") print("Sentiment refresh complete.\n") # ============================================================ # API ENDPOINTS — Sentiment # ============================================================ @app.get("/api/sentiment") def get_all_sentiments(): return [get_sentiment_score(sym) for sym in SYMBOLS] @app.get("/api/sentiment/{symbol}") def get_single_sentiment(symbol: str): return get_sentiment_score(symbol.upper()) @app.get("/api/history/{symbol}") def get_symbol_history(symbol: str, days: int = 7): return {"symbol": symbol.upper(), "history": get_history(symbol.upper(), days)} @app.get("/api/option-chain/{symbol}") def get_option_chain(symbol: str): return fetch_option_chain_levels(symbol.upper()) @app.get("/api/constituents/{index}") def get_constituents(index: str): index_upper = index.upper() if index_upper not in ("NIFTY", "SENSEX"): return {"error": "Index must be NIFTY or SENSEX"} return fetch_constituents(index_upper) @app.get("/api/action-dashboard") def get_action_dashboard(): try: events = generate_action_events() return { "updated": datetime.now(IST).strftime("%d %b %Y, %I:%M %p IST"), "count": len(events), "max_events": ACTION_DASHBOARD_MAX_EVENTS, "events": events } except Exception as e: print(f"⚠️ Action dashboard error: {e}") return {"updated": datetime.now(IST).strftime("%d %b %Y, %I:%M %p IST"), "count": 0, "events": [], "error": str(e)[:200]} @app.get("/api/range") def get_expected_range(): cached = cache_get("range") if cached: return cached try: nifty = yf.Ticker("^NSEI") vix = yf.Ticker("^INDIAVIX") nifty_data = nifty.history(period="1d") vix_data = vix.history(period="1d") if nifty_data.empty or vix_data.empty: return {"error": "Could not fetch market data."} last_close = nifty_data['Close'].iloc[-1] current_vix = vix_data['Close'].iloc[-1] vix_decimal = current_vix / 100 if current_vix > 1.0 else current_vix daily_move = last_close * (vix_decimal / math.sqrt(252)) result = { "last_close": round(last_close, 2), "vix": round(current_vix, 2), "low": round(last_close - daily_move, 2), "high": round(last_close + daily_move, 2), "mid": round(last_close, 2) } cache_set("range", result) return result except Exception as e: return {"error": str(e)} # ============================================================ # API ENDPOINTS — Zone Scanner # ============================================================ @app.get("/api/scan/zones-15m") def get_zone_scan(): cached = cache_get("latest_scan") if cached: return cached return run_zone_scan() @app.get("/api/scan/rescan") def force_zone_rescan(min_price: float = None, max_price: float = None): return run_zone_scan(min_price=min_price, max_price=max_price) @app.get("/api/scan/active-zones") def get_active_zones_endpoint(): zones = get_active_zones() return {"count": len(zones), "zones": zones} @app.get("/api/scan/recent-events") def get_recent_zone_events(hours: int = 24): cutoff = (datetime.now(IST) - timedelta(hours=hours)).isoformat() conn = sqlite3.connect(ZONES_DB_PATH) c = conn.cursor() c.execute(""" SELECT symbol, name, sector, event_type, zone_type, zone_top, zone_bottom, price, event_at, tier, message FROM zone_events WHERE event_at >= ? ORDER BY event_at DESC LIMIT 100 """, (cutoff,)) rows = c.fetchall() conn.close() return { "count": len(rows), "events": [{ "symbol": r[0], "name": r[1], "sector": r[2], "event_type": r[3], "zone_type": r[4], "zone_top": r[5], "zone_bottom": r[6], "price": r[7], "event_at": r[8], "tier": r[9], "message": r[10], } for r in rows] } # ============================================================ # API ENDPOINTS — Shared # ============================================================ @app.get("/api/status") def get_market_status(): now = datetime.now(IST) is_weekday = now.weekday() < 5 is_market_hours = 9 <= now.hour < 15 or (now.hour == 15 and now.minute <= 30) status = "Market Open" if (is_weekday and is_market_hours) else "Market Closed" return {"status": status, "updated": now.strftime("%d %b %Y, %I:%M %p IST"), "version": VERSION} @app.get("/api/scan/status") def scan_status(min_price: float = None, max_price: float = None): now = datetime.now(IST) is_weekday = now.weekday() < 5 is_market_hours = 9 <= now.hour < 15 or (now.hour == 15 and now.minute <= 30) status = "Market Open" if (is_weekday and is_market_hours) else "Market Closed" _min = min_price if min_price is not None else MIN_PRICE _max = max_price if max_price is not None else MAX_PRICE return { "version": VERSION, "status": status, "updated": now.strftime("%d %b %Y, %I:%M %p IST"), "universe_count": len(get_fno_stocks()), "active_zones": len(get_active_zones()), "price_filter": f"₹{int(_min)} - ₹{int(_max)}", "price_min": _min, "price_max": _max, "timeframe": "15m", "pivot_config": {"left": PIVOT_LEFT, "right": PIVOT_RIGHT, "zone_length_bars": ZONE_LENGTH_BARS} } @app.get("/api/version") def get_version(): return { "version": VERSION, "date": "2026-10-02", "features": [ "UNIFIED PLATFORM v1.0.0", "--- Sentiment Scanner ---", "Multi-source AI sentiment (News + Twitter + YouTube)", "FII/DII institutional flows", "Option chain (Support/Resistance/Max Pain/PCR)", "OI Buildup with net writer bias", "Nifty 50 Top 25 + Sensex 30 constituents", "Weight-aware + ATR-based tiered alerting", "Live Action Dashboard", "--- Zone Scanner ---", "15-minute reversal zone scanner", "Pine Script pivot detection logic", "F&O stock universe (~190 stocks)", "New zone + zone entry detection", "Tiered alerts (IMPORTANT / SIGNAL)", "Automatic 15-minute scan schedule", "--- Shared ---", "Adaptive caching (market-hours-aware)", "SQLite persistence (2 DBs)", "Scheduled snapshots" ] } @app.get("/api/scheduler/status") def scheduler_status(): jobs = scheduler.get_jobs() return { "version": VERSION, "timezone": str(IST), "current_time": datetime.now(IST).strftime("%Y-%m-%d %H:%M:%S IST"), "market_hours": is_market_hours_now(), "jobs": [ {"id": j.id, "name": j.name, "next_run": j.next_run_time.strftime("%Y-%m-%d %H:%M:%S %Z") if j.next_run_time else None} for j in jobs ] } # ============================================================ # SCHEDULER SETUP # ============================================================ scheduler = BackgroundScheduler(timezone=IST) # Sentiment: 9:00, 12:30, 15:30 IST scheduler.add_job(scheduled_sentiment_refresh, trigger=CronTrigger(day_of_week='mon-fri', hour=9, minute=0, timezone=IST), id='sentiment_9am', name='Sentiment 9:00 AM', replace_existing=True) scheduler.add_job(scheduled_sentiment_refresh, trigger=CronTrigger(day_of_week='mon-fri', hour=12, minute=30, timezone=IST), id='sentiment_1230pm', name='Sentiment 12:30 PM', replace_existing=True) scheduler.add_job(scheduled_sentiment_refresh, trigger=CronTrigger(day_of_week='mon-fri', hour=15, minute=30, timezone=IST), id='sentiment_330pm', name='Sentiment 3:30 PM', replace_existing=True) # Zone scan: every 15 min from 9:30-15:30, + final 16:30 for hour in range(9, 16): for minute in [0, 15, 30, 45]: if hour == 9 and minute < 30: continue if hour == 15 and minute > 30: continue try: scheduler.add_job(run_zone_scan, trigger=CronTrigger(day_of_week="mon-fri", hour=hour, minute=minute, second=2, timezone=IST), id=f"zone_{hour:02d}{minute:02d}", name=f"Zone scan {hour:02d}:{minute:02d}", replace_existing=True, misfire_grace_time=120) except Exception: pass scheduler.add_job(run_zone_scan, trigger=CronTrigger(day_of_week="mon-fri", hour=16, minute=30, second=2, timezone=IST), id="zone_1630_final", name="Final zone scan 16:30", replace_existing=True, misfire_grace_time=600) scheduler.start() print(f"\n🚀 PDMP Unified Platform {VERSION}") print(f" Sentiment: 3 daily snapshots (9:00 / 12:30 / 15:30 IST)") print(f" Zone Scanner: every 15 min + final 16:30 IST") print(f" Universe: F&O (~190 stocks) + Sentiment (6 symbols)") print(f" API: http://0.0.0.0:8000") print(f" Both scanners running from one process\n") # ============================================================ # BACKFILL ZONES ON STARTUP # ============================================================ def delayed_backfill(): time.sleep(3) try: print("Running initial zone backfill...") run_zone_scan() print("Zone backfill complete.\n") except Exception as e: print(f"Zone backfill error: {e}") threading.Thread(target=delayed_backfill, daemon=True).start() # ============================================================ # ENTRY POINT # ============================================================ if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)