import uuid import datetime from typing import List, Optional from sqlalchemy import select, update, delete, and_, func from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.dialects.postgresql import insert as pg_insert from backend.app.database import models from backend.app.schemas import schemas # ========================================== # USER OPERATIONS # ========================================== async def get_user(db: AsyncSession, user_id: uuid.UUID) -> Optional[models.User]: result = await db.execute(select(models.User).where(models.User.id == user_id)) return result.scalars().first() async def get_user_by_email(db: AsyncSession, email: str) -> Optional[models.User]: result = await db.execute(select(models.User).where(models.User.email == email)) return result.scalars().first() async def get_user_by_google_id(db: AsyncSession, google_id: str) -> Optional[models.User]: result = await db.execute(select(models.User).where(models.User.google_id == google_id)) return result.scalars().first() async def create_user(db: AsyncSession, user_in: schemas.UserBase, google_id: str) -> models.User: db_user = models.User( email=user_in.email, full_name=user_in.full_name, picture_url=user_in.picture_url, google_id=google_id, credits=5, # Starting free credits last_credit_refresh=datetime.datetime.now(datetime.timezone.utc) ) db.add(db_user) await db.commit() await db.refresh(db_user) return db_user async def refresh_user_credits(db: AsyncSession, user: models.User) -> models.User: """ Checks and applies user refreshes: - For Free tier: resets credits to 5 every 7 days. - For Pro tier: resets monthly_messages_used to 0 every 30 days. """ now = datetime.datetime.now(datetime.timezone.utc) updated = False if user.subscription_tier == "free": time_elapsed = now - user.last_credit_refresh if time_elapsed >= datetime.timedelta(days=7): user.credits = 5 user.last_credit_refresh = now updated = True if user.subscription_tier == "pro": billing_elapsed = now - user.last_billing_date if billing_elapsed >= datetime.timedelta(days=30): user.monthly_messages_used = 0 user.last_billing_date = now updated = True if updated: db.add(user) await db.commit() await db.refresh(user) return user async def deduct_user_credit(db: AsyncSession, user_id: uuid.UUID) -> bool: """ Deducts 1 credit from the user's account. Returns True if deduction succeeded, False if user has 0 credits. """ user = await get_user(db, user_id) if not user: return False if user.email == "karanshelar8775@gmail.com": return True if user.credits <= 0: return False user.credits -= 1 db.add(user) await db.commit() return True # ========================================== # WATCHLIST OPERATIONS # ========================================== async def get_user_watchlist(db: AsyncSession, user_id: uuid.UUID) -> List[models.Watchlist]: # Single query ordered by creation time (oldest first so sidebar shows consistent ordering) result = await db.execute( select(models.Watchlist) .where(models.Watchlist.user_id == user_id) .order_by(models.Watchlist.created_at.asc()) ) return list(result.scalars().all()) async def add_to_watchlist(db: AsyncSession, user_id: uuid.UUID, ticker: str) -> models.Watchlist: # Check if already exists to prevent duplicate entries existing_result = await db.execute( select(models.Watchlist).where( and_(models.Watchlist.user_id == user_id, models.Watchlist.ticker == ticker.upper()) ) ) # Store the scalar before the cursor is consumed; calling .first() twice on the # same cursor always returns None on the second call (SQLAlchemy cursor exhaustion). existing_item = existing_result.scalars().first() if existing_item: return existing_item db_watchlist = models.Watchlist(user_id=user_id, ticker=ticker.upper()) db.add(db_watchlist) await db.commit() await db.refresh(db_watchlist) return db_watchlist async def remove_from_watchlist(db: AsyncSession, user_id: uuid.UUID, ticker: str) -> bool: result = await db.execute( delete(models.Watchlist).where( and_(models.Watchlist.user_id == user_id, models.Watchlist.ticker == ticker.upper()) ) ) await db.commit() return result.rowcount > 0 # ========================================== # ALERT OPERATIONS # ========================================== async def get_user_alerts(db: AsyncSession, user_id: uuid.UUID) -> List[models.Alert]: result = await db.execute( select(models.Alert) .where(models.Alert.user_id == user_id) .order_by(models.Alert.created_at.desc()) ) return list(result.scalars().all()) async def get_active_alerts(db: AsyncSession) -> List[models.Alert]: result = await db.execute( select(models.Alert).where(models.Alert.is_active == True) ) return list(result.scalars().all()) async def create_alert(db: AsyncSession, user_id: uuid.UUID, alert_in: schemas.AlertCreate) -> models.Alert: db_alert = models.Alert( user_id=user_id, ticker=alert_in.ticker.upper(), target_price=alert_in.target_price, condition=alert_in.condition.lower(), is_active=True ) db.add(db_alert) await db.commit() await db.refresh(db_alert) return db_alert async def deactivate_alert(db: AsyncSession, alert_id: uuid.UUID) -> bool: result = await db.execute( update(models.Alert) .where(models.Alert.id == alert_id) .values(is_active=False) ) await db.commit() return result.rowcount > 0 async def trigger_alert(db: AsyncSession, alert_id: uuid.UUID) -> bool: result = await db.execute( update(models.Alert) .where(models.Alert.id == alert_id) .values(is_triggered=True, last_notified_at=func.now()) ) await db.commit() return result.rowcount > 0 async def update_alert_notification_time(db: AsyncSession, alert_id: uuid.UUID) -> bool: result = await db.execute( update(models.Alert) .where(models.Alert.id == alert_id) .values(last_notified_at=func.now()) ) await db.commit() return result.rowcount > 0 # ========================================== # STOCK HISTORY OPERATIONS # ========================================== async def get_stock_history(db: AsyncSession, ticker: str, limit: int = 100) -> List[models.StockHistory]: result = await db.execute( select(models.StockHistory) .where(models.StockHistory.ticker == ticker.upper()) .order_by(models.StockHistory.timestamp.desc()) .limit(limit) ) history = list(result.scalars().all()) # If database has under 100 records for this ticker, dynamically backfill from yfinance. # We use a 15-minute Redis-based lockout to avoid hammering yfinance when database is cold. if len(history) < 100: import redis from backend.app.config.settings import settings # Connect to Redis logical DB 1 (same as service cache) redis_client = None if settings.REDIS_URL: try: redis_cache_url = settings.REDIS_URL if redis_cache_url.endswith("/0"): redis_cache_url = redis_cache_url[:-2] + "/1" elif not any(redis_cache_url.endswith(f"/{i}") for i in range(16)): redis_cache_url = redis_cache_url.rstrip("/") + "/1" redis_client = redis.from_url(redis_cache_url, decode_responses=True) except Exception: pass lock_key = f"quantiq:backfill_lock:{ticker.upper()}" already_attempted = False if redis_client: try: already_attempted = bool(redis_client.get(lock_key)) except Exception: pass if not already_attempted: if redis_client: try: redis_client.setex(lock_key, 900, "1") # 15 minutes TTL except Exception: pass import yfinance as yf import asyncio try: # yfinance allows fetching 1m interval historical data up to 30 days. We fetch last 5 days. yf_ticker = yf.Ticker(ticker.upper()) df = await asyncio.get_event_loop().run_in_executor( None, lambda: yf_ticker.history(period="5d", interval="1m") ) if df is not None and not df.empty: df = df.reset_index() # Identify timestamp column time_col = None for col in ['Date', 'Datetime', 'index', 'timestamp']: if col in df.columns: time_col = col break if time_col: candles_to_insert = [] for _, row in df.iterrows(): ts = row[time_col] if hasattr(ts, 'to_pydatetime'): ts_dt = ts.to_pydatetime() elif isinstance(ts, str): ts_dt = datetime.datetime.fromisoformat(ts) else: ts_dt = ts if ts_dt.tzinfo is not None: ts_dt = ts_dt.replace(tzinfo=None) candles_to_insert.append({ "ticker": ticker.upper(), "timestamp": ts_dt, "open": float(row["Open"]), "high": float(row["High"]), "low": float(row["Low"]), "close": float(row["Close"]), "volume": int(row["Volume"]) if "Volume" in row else 0 }) if candles_to_insert: # Perform batch insert ignoring duplicates stmt = pg_insert(models.StockHistory).values(candles_to_insert) await db.execute(stmt.on_conflict_do_nothing(index_elements=["ticker", "timestamp"])) await db.commit() # Re-query the database with the fully backfilled candles result = await db.execute( select(models.StockHistory) .where(models.StockHistory.ticker == ticker.upper()) .order_by(models.StockHistory.timestamp.desc()) .limit(limit) ) history = list(result.scalars().all()) except Exception as e: print(f"MLOps Dynamic Backfill: Failed to populate history for {ticker}: {e}") # Return in chronological order (oldest to newest) for indicators history.reverse() return history async def insert_stock_candle(db: AsyncSession, candle: schemas.StockHistoryBase) -> None: """ Inserts a single candle record. If it already exists, do nothing. """ stmt = pg_insert(models.StockHistory).values( ticker=candle.ticker.upper(), timestamp=candle.timestamp, open=candle.open, high=candle.high, low=candle.low, close=candle.close, volume=candle.volume ) # PostgreSQL specific upsert: do nothing on conflict stmt = stmt.on_conflict_do_nothing(index_elements=["ticker", "timestamp"]) await db.execute(stmt) await db.commit() # Check and trigger alerts for this ticker alert_stmt = ( select(models.Alert) .where(models.Alert.ticker == candle.ticker.upper()) .where(models.Alert.is_active == True) .where(models.Alert.is_triggered == False) ) alert_result = await db.execute(alert_stmt) active_alerts = list(alert_result.scalars().all()) from backend.app.services.email_service import send_price_alert_email for alert in active_alerts: triggered = False if alert.condition == "above" and candle.close >= alert.target_price: triggered = True elif alert.condition == "below" and candle.close <= alert.target_price: triggered = True if triggered: alert.is_triggered = True alert.last_notified_at = func.now() db.add(alert) await db.commit() # Retrieve user to get their email address user = await get_user(db, alert.user_id) if user: send_price_alert_email( to_email=user.email, ticker=alert.ticker, condition=alert.condition, target_price=alert.target_price, current_price=candle.close ) # ========================================== # PAYMENT TRANSACTION OPERATIONS # ========================================== async def create_payment_transaction( db: AsyncSession, user_id: uuid.UUID, order_id: str, amount: int, credits_credited: int ) -> models.PaymentTransaction: db_tx = models.PaymentTransaction( user_id=user_id, razorpay_order_id=order_id, amount=amount, status="created", credits_credited=credits_credited ) db.add(db_tx) await db.commit() await db.refresh(db_tx) return db_tx async def capture_payment_transaction( db: AsyncSession, order_id: str, payment_id: str ) -> Optional[models.PaymentTransaction]: """ Captures a pending transaction, updates status, and credits the user. Uses a transaction block to ensure atomic operations. """ # 1. Fetch transaction result = await db.execute( select(models.PaymentTransaction).where(models.PaymentTransaction.razorpay_order_id == order_id) ) tx = result.scalars().first() if not tx or tx.status == "captured": return tx # 2. Update transaction tx.razorpay_payment_id = payment_id tx.status = "captured" db.add(tx) # 3. Credit the user and update subscription tier user = await get_user(db, tx.user_id) if user: user.credits += tx.credits_credited # Determine plan from transaction amount in Rupees amt_rupees = tx.amount // 100 if amt_rupees == 500: user.subscription_tier = "analyst" user.messages_remaining += 10 elif amt_rupees == 1500: user.subscription_tier = "trader" user.messages_remaining += 25 elif amt_rupees in (10000, 15000): user.subscription_tier = "pro" user.monthly_messages_used = 0 user.last_billing_date = datetime.datetime.now(datetime.timezone.utc) db.add(user) await db.commit() await db.refresh(tx) return tx async def create_saved_strategy( db: AsyncSession, user_id: uuid.UUID, ticker: str, bullish_probability: int, reason: str ) -> models.SavedStrategy: db_strategy = models.SavedStrategy( user_id=user_id, ticker=ticker.upper(), bullish_probability=bullish_probability, reason=reason ) db.add(db_strategy) await db.commit() await db.refresh(db_strategy) return db_strategy async def get_user_saved_strategies(db: AsyncSession, user_id: uuid.UUID) -> List[models.SavedStrategy]: result = await db.execute( select(models.SavedStrategy) .where(models.SavedStrategy.user_id == user_id) .order_by(models.SavedStrategy.created_at.desc()) ) return list(result.scalars().all()) # ========================================== # MLOPS PREDICTION LOG OPERATIONS # ========================================== async def create_prediction_log( db: AsyncSession, user_id: uuid.UUID, ticker: str, model_version: str, confidence: float, predicted_action: str, entry_price: float, target_price: Optional[float] = None, stop_loss: Optional[float] = None, asset_class: Optional[str] = None ) -> models.PredictionLog: db_log = models.PredictionLog( user_id=user_id, ticker=ticker.upper(), model_version=model_version, confidence=confidence, predicted_action=predicted_action, entry_price=entry_price, target_price=target_price, stop_loss=stop_loss, asset_class=asset_class, status="pending" ) db.add(db_log) await db.commit() await db.refresh(db_log) return db_log async def get_pending_predictions(db: AsyncSession) -> List[models.PredictionLog]: result = await db.execute( select(models.PredictionLog) .where(models.PredictionLog.status == "pending") .order_by(models.PredictionLog.timestamp.asc()) ) return list(result.scalars().all()) async def get_all_prediction_logs(db: AsyncSession, limit: int = 1000) -> List[models.PredictionLog]: result = await db.execute( select(models.PredictionLog) .order_by(models.PredictionLog.timestamp.desc()) .limit(limit) ) return list(result.scalars().all()) # ========================================== # MLOPS STRATEGY LOG OPERATIONS # ========================================== async def create_strategy_log( db: AsyncSession, user_id: uuid.UUID, ticker: str, model_version: str, bullish_probability: int, ai_entry: float, ai_target: float, ai_stop_loss: float, user_entry: float, user_target: float, user_stop_loss: float, asset_class: Optional[str] = None ) -> models.StrategyLog: db_strategy = models.StrategyLog( user_id=user_id, ticker=ticker.upper(), model_version=model_version, bullish_probability=bullish_probability, ai_entry=ai_entry, ai_target=ai_target, ai_stop_loss=ai_stop_loss, user_entry=user_entry, user_target=user_target, user_stop_loss=user_stop_loss, asset_class=asset_class, status="pending" ) db.add(db_strategy) await db.commit() await db.refresh(db_strategy) return db_strategy async def get_pending_strategy_logs(db: AsyncSession) -> List[models.StrategyLog]: result = await db.execute( select(models.StrategyLog) .where(models.StrategyLog.status == "pending") .order_by(models.StrategyLog.timestamp.asc()) ) return list(result.scalars().all()) async def get_all_strategy_logs(db: AsyncSession, limit: int = 1000) -> List[models.StrategyLog]: result = await db.execute( select(models.StrategyLog) .order_by(models.StrategyLog.timestamp.desc()) .limit(limit) ) return list(result.scalars().all()) async def get_recently_analyzed_tickers(db: AsyncSession, user_id: uuid.UUID, limit: int = 5) -> List[str]: """ Retrieves the unique tickers that the specified user has analyzed recently. """ from sqlalchemy import func result = await db.execute( select(models.PredictionLog.ticker, func.max(models.PredictionLog.timestamp).label("latest")) .where(models.PredictionLog.user_id == user_id) .group_by(models.PredictionLog.ticker) .order_by(func.max(models.PredictionLog.timestamp).desc()) .limit(limit) ) return [row.ticker for row in result] async def get_trending_tickers(db: AsyncSession, limit: int = 5) -> List[dict]: """ Retrieves the globally trending tickers based on analysis query count in the last 7 days. """ from sqlalchemy import func cutoff = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=7) result = await db.execute( select(models.PredictionLog.ticker, func.count(models.PredictionLog.id).label("count")) .where(models.PredictionLog.timestamp >= cutoff) .group_by(models.PredictionLog.ticker) .order_by(func.count(models.PredictionLog.id).desc()) .limit(limit) ) return [{"ticker": row.ticker, "count": row.count} for row in result]