# app/v1/services/gpt_prompts.py """Prompt templates and builders for GPT-based trading analysis. This module is intentionally focused only on prompt construction so that you can grow and tweak prompts independently of the GPT client logic in `gpt_engine.py`. """ from typing import Dict, Any, Optional import os def _strip_inline_comment(v: Optional[str]) -> str: """Strip inline `.env` comments like `value # comment` or `value#comment`.""" if v is None: return "" s = str(v) if "#" in s: s = s.split("#", 1)[0] return s.strip() def _env_prompt_mode() -> str: mode = _strip_inline_comment(os.getenv("GPT_PROMPT_MODE", "compact") or "compact").lower() return mode if mode in {"full", "compact"} else "full" def _env_int(name: str, default: int, *, min_value: int = 0, max_value: int = 10_000) -> int: raw = _strip_inline_comment(os.getenv(name)) if raw == "": return int(default) try: v = int(raw) except Exception: return int(default) return int(max(min_value, min(max_value, v))) def _safe_float(v): try: if v is None: return None return float(v) except Exception: return None def _compact_market_intelligence(market_intelligence: Optional[Dict[str, Any]]) -> str: """Render a compact broad-market context block for the prompt. This is intentionally best-effort and schema-tolerant because the background job may evolve over time. """ if not isinstance(market_intelligence, dict) or not market_intelligence: return "" # Keep this small; only include the most actionable keys. out: Dict[str, Any] = {} for k in ( "as_of", "market_bias", "risk_regime", "volatility_regime", "notes", "top_sectors", "sector_strength", # Per-symbol sector/industry context (best-effort) "symbol_sector", "symbol_industry", "symbol_sector_trend", ): if k in market_intelligence and market_intelligence.get(k) not in (None, ""): out[k] = market_intelligence.get(k) if not out: return "" # Ensure this stays one short JSON-like object. try: import json rendered = json.dumps(out, ensure_ascii=False) except Exception: rendered = str(out) return f"\nMARKET INTELLIGENCE (broad market context):\n{rendered}\n" def _compact_market_context(symbol: str, market_data: Dict[str, Any]) -> str: """Render a compact, token-efficient market snapshot. Intentionally avoids raw candle series. Includes: - Quote summary - Last candle + last-close change per timeframe (if available) - Latest indicator snapshot per timeframe - Strategy flags - Key levels (pivots / fib zone if present) """ lines = [] lines.append(f"SYMBOL: {symbol}") include_quote = _env_int("GPT_COMPACT_INCLUDE_QUOTE", 0, min_value=0, max_value=1) == 1 quote = market_data.get("quote") or {} if include_quote and quote: ohlc = quote.get("ohlc") or {} lines.append( "QUOTE: " f"ltp={quote.get('last_price')} " f"o={ohlc.get('open')} h={ohlc.get('high')} l={ohlc.get('low')} prev_c={ohlc.get('close')} " f"vol={quote.get('volume')} chg={quote.get('net_change')} ({quote.get('net_change_percentage')}%)" ) # Default to including last-bar summaries; it materially improves GPT decisions with minimal token cost. include_last_bar = _env_int("GPT_COMPACT_INCLUDE_TF_LAST_BAR", 1, min_value=0, max_value=1) == 1 candles = market_data.get("candles") or {} if include_last_bar and isinstance(candles, dict) and candles: lines.append("TF_LAST_BAR:") for tf, series in candles.items(): if not series: continue last = series[-1] or {} prev = series[-2] if len(series) >= 2 else None last_close = _safe_float(last.get("close")) prev_close = _safe_float(prev.get("close")) if isinstance(prev, dict) else None chg_pct = None if last_close is not None and prev_close and prev_close != 0: chg_pct = (last_close - prev_close) / prev_close * 100.0 o = last.get("open") h = last.get("high") l = last.get("low") c = last.get("close") v = last.get("volume") chg = f" chg%={chg_pct:.2f}" if chg_pct is not None else "" lines.append(f"- {tf}: O={o} H={h} L={l} C={c} V={v}{chg}") indicators = market_data.get("indicators") or {} if isinstance(indicators, dict) and indicators: lines.append("INDICATORS (latest):") for tf, summary in indicators.items(): if not summary: continue ema_values = (summary.get("ema") or {}) if isinstance(summary, dict) else {} ema_block = "" if isinstance(ema_values, dict) and ema_values: try: ema_items = [f"EMA{p}={ema_values[p]}" for p in sorted(ema_values.keys(), key=lambda x: int(x))] except Exception: ema_items = [f"EMA{k}={v}" for k, v in ema_values.items()] ema_block = " | " + ", ".join(ema_items) lines.append( f"- {tf}: close={summary.get('close')} vwap={summary.get('vwap')} rsi={summary.get('rsi')} " f"macd={summary.get('macd')}/{summary.get('macd_signal')}/{summary.get('macd_hist')} " f"vol={summary.get('volume')} dvol={summary.get('volume_delta')}{ema_block}" ) include_pivots = _env_int("GPT_COMPACT_INCLUDE_PIVOTS", 1, min_value=0, max_value=1) == 1 pivots = market_data.get("pivots") or {} if include_pivots and isinstance(pivots, dict) and pivots: lines.append("PIVOTS:") for tf, levels in pivots.items(): if not levels: continue lines.append( f"- {tf}: P={levels.get('P')} R1={levels.get('R1')} S1={levels.get('S1')}" ) # Only include fib zone (not full retracements dict) if available via strategies. strategies = market_data.get("strategies") or {} if isinstance(strategies, dict) and strategies: per_tf = strategies.get("per_timeframe") or {} mt = strategies.get("multi_timeframe") or {} lines.append("STRATEGY_FLAGS:") if isinstance(per_tf, dict): for tf, feats in per_tf.items(): if not feats: continue trend = feats.get("trend") or {} vol = feats.get("volatility") or {} rb = feats.get("range_breakout") or {} hod = feats.get("hod_lod") or {} mb = feats.get("momentum_burst") or {} vc = feats.get("volume_climax") or {} pat = feats.get("pattern") or {} fib_zone = feats.get("fib_zone") or {} lines.append( f"- {tf}: trend={trend.get('direction')}({trend.get('strength')}) " f"vol_regime={vol.get('regime')} rb={rb.get('status')} " f"near_hod={hod.get('near_hod')} near_lod={hod.get('near_lod')} " f"mom={mb.get('status')} vr={mb.get('volume_ratio')} " f"climax={vc.get('climax')} vr={vc.get('volume_ratio')} " f"pattern={pat.get('pattern')} fib_zone={fib_zone.get('zone')}" ) if mt: lines.append(f"MTF: trend_alignment={mt.get('trend_alignment')}") return "\n".join(lines) def _prepare_market_data_prompt_compact( symbol: str, market_data: Dict[str, Any], question: str, context: str, market_intelligence: Optional[Dict[str, Any]] = None, ) -> str: user_question = (question or "Provide intraday trading analysis").strip() execution_context = (context or "NEUTRAL").strip().upper() or "NEUTRAL" prompt = f"""SYSTEM ROLE: You are an expert institutional intraday trader and risk manager. You do NOT follow fixed strategies. You infer the best intraday decision purely from the data provided. Priority: 1) Capital protection 2) Trade only when edge is clear 3) Precision over frequency If no clear edge exists, you MUST choose HOLD. EXECUTION CONTEXT (informational only, NOT a rule): {execution_context} You MUST ignore this context if price action or data contradicts it. SYMBOL: {symbol} USER QUESTION: {user_question} MARKET SNAPSHOT: {_compact_market_context(symbol, market_data)} {_compact_market_intelligence(market_intelligence)} ANALYZE (full depth): 1) Determine intraday regime (trending/ranging/transitioning/volatile-uncertain) 2) Assess direction bias (bullish/bearish/neutral) 3) Decide if there is REAL intraday edge; if weak/unclear => HOLD 4) If edge exists: define entry zone, stop-loss ZONE, and TWO intraday targets 5) Identify early_breakout only if range/key level break + strong momentum/volume DATA DISCIPLINE: - Use ONLY data provided. NEVER invent missing candles/levels/indicators. - If raw candles are absent, rely ONLY on computed indicators/flags. SCORING (0-100): 0-40 HOLD/no-trade; 41-59 weak; 60-74 acceptable; 75-84 strong; 85-100 exceptional. CONVICTION MAPPING (mandatory): - VERY_STRONG => score >= 85 - HIGH => score 75-84 - MEDIUM => score 60-74 - LOW => score < 60 HOLD can NEVER be VERY_STRONG or HIGH. STOP-LOSS AS A ZONE (mandatory): BUY: stop_loss_zone.upper near invalidation, stop_loss_zone.lower hard invalidation SELL: stop_loss_zone.lower near invalidation, stop_loss_zone.upper hard invalidation OUTPUT FORMAT (STRICT JSON ONLY): Return ONE valid JSON object with EXACTLY these keys: {{ "decision": "BUY"|"SELL"|"HOLD", "conviction": "VERY_STRONG"|"HIGH"|"MEDIUM"|"LOW", "trend_label": "", "entry_price": null, "entry_zone": null, "stop_loss": null, "stop_loss_zone": null, "targets": [], "score": 0, "risk_profile": "aggressive"|"conservative", "early_breakout": false, "early_breakout_comment": "", "rationale": ["..."], "technical_indicators": {{ "trend": "bullish"|"bearish"|"sideways", "momentum": "strong"|"weak"|"neutral", "volume": "high"|"normal"|"low", "support": 0.0, "resistance": 0.0, "accumulation_distribution": "accumulation"|"distribution"|"neutral", "pattern_summary": "" }}, "risk_reward_ratio": 0.0, "time_horizon": "intraday", "notes": "" }} NON-NEGOTIABLE: - Output ONLY valid JSON. No code blocks. - If decision=HOLD: entry_price=null, entry_zone=null, stop_loss=null, stop_loss_zone=null, targets=[], score<40, conviction=LOW. """ return prompt def prepare_market_data_prompt( symbol: str, market_data: Dict[str, Any], question: str, context: str, market_intelligence: Optional[Dict[str, Any]] = None, ) -> str: """Build a rich, strictly-structured prompt for intraday analysis. We push *all* heavy lifting to GPT here: trend reading, early breakouts, pseudo-indicators, scoring and the final BUY/SELL/HOLD decision. Downstream code just consumes the JSON. The "context" parameter selects the ACTIVE STRATEGY for this request. For now we primarily support intraday SHORT_SELL and LONG_BUY daily trades, but the structure is future-proof for additional strategies. """ # Token-cost control: keep existing behavior unless explicitly switched. if _env_prompt_mode() == "compact": return _prepare_market_data_prompt_compact(symbol, market_data, question, context, market_intelligence) user_question = (question or "Provide intraday trading analysis").strip() execution_context = (context or "NEUTRAL").strip().upper() or "NEUTRAL" prompt = f"""SYSTEM ROLE: You are an expert institutional intraday trader and risk manager. You do NOT follow fixed strategies. You infer the best intraday decision purely from the data provided. Priority: 1) Capital protection 2) Trade only when edge is clear 3) Precision over frequency If no clear edge exists, you MUST choose HOLD. EXECUTION CONTEXT (informational only, NOT a rule): {execution_context} You MUST ignore this context if price action or data contradicts it. SYMBOL: {symbol} USER QUESTION: {user_question} {_compact_market_intelligence(market_intelligence)} IMPORTANT: - If raw candle data is present below, you MUST use it. - If raw candle data is NOT present, rely ONLY on computed indicators/flags. - NEVER invent missing data. ANALYZE (full depth): 1) Determine intraday regime (trending/ranging/transitioning/volatile-uncertain) 2) Assess direction bias (bullish/bearish/neutral) 3) Decide if there is REAL intraday edge; if weak/unclear => HOLD 4) If edge exists: define entry zone, stop-loss ZONE, and TWO intraday targets 5) Identify early_breakout only if range/key level break + strong momentum/volume STOP-LOSS AS A ZONE (mandatory): BUY: stop_loss_zone.upper near invalidation, stop_loss_zone.lower hard invalidation SELL: stop_loss_zone.lower near invalidation, stop_loss_zone.upper hard invalidation RAW MARKET SNAPSHOT (may include quote + raw candles): """ # Add quote data quote = market_data.get("quote", {}) if quote: prompt += f""" CURRENT QUOTE (may be delayed slightly): - Last Price: ₹{quote.get('last_price', 'N/A')} - Open: ₹{quote.get('ohlc', {}).get('open', 'N/A')} - High: ₹{quote.get('ohlc', {}).get('high', 'N/A')} - Low: ₹{quote.get('ohlc', {}).get('low', 'N/A')} - Close (prev): ₹{quote.get('ohlc', {}).get('close', 'N/A')} - Volume: {quote.get('volume', 'N/A')} - Change: {quote.get('net_change', 'N/A')} ({quote.get('net_change_percentage', 'N/A')}%) """ # Add candle data summary per timeframe # Cost control: this is the biggest token driver in FULL mode. # Set GPT_CANDLES_PER_TF=0 to disable dumping raw candle lines. candles_per_tf = _env_int("GPT_CANDLES_PER_TF", 6, min_value=0, max_value=50) candles = market_data.get("candles", {}) for timeframe, candle_data in candles.items(): if candle_data and len(candle_data) > 0 and candles_per_tf > 0: recent_candles = candle_data[-candles_per_tf:] prompt += f""" TIMEFRAME: {timeframe.upper()} (Last {len(recent_candles)} candles) Format per line: index: O,H,L,C,Vol """ for i, candle in enumerate(recent_candles): o = candle.get("open") h = candle.get("high") l = candle.get("low") c = candle.get("close") v = candle.get("volume", 0) prompt += f" {i+1}: {o}, {h}, {l}, {c}, {v}\n" # Add numeric indicator summaries if available indicators = market_data.get("indicators", {}) if indicators: prompt += """ --- INDICATOR SNAPSHOT (per timeframe, latest bar) """ for timeframe, summary in indicators.items(): ema_block = "" ema_values = summary.get("ema", {}) or {} if ema_values: # Compact EMA listing sorted by period ema_items = [f"EMA{p}={v}" for p, v in sorted(ema_values.items(), key=lambda x: int(x[0]))] ema_block = ", ".join(ema_items) prompt += f""" TIMEFRAME: {timeframe.upper()} INDICATORS - Close: {summary.get('close')} - VWAP: {summary.get('vwap')} - RSI: {summary.get('rsi')} - MACD: {summary.get('macd')} / Signal: {summary.get('macd_signal')} / Hist: {summary.get('macd_hist')} - Volume: {summary.get('volume')} (Δ {summary.get('volume_delta')}) - EMAs: {ema_block} """ # Add pivot levels if present (currently daily) pivots = market_data.get("pivots", {}) if pivots: prompt += """ --- PIVOT LEVELS (supports/resistances from higher timeframe) """ for tf, levels in pivots.items(): if not levels: continue prompt += f""" TIMEFRAME: {tf.upper()} PIVOTS - P: {levels.get('P')} - R1/R2/R3: {levels.get('R1')}, {levels.get('R2')}, {levels.get('R3')} - S1/S2/S3: {levels.get('S1')}, {levels.get('S2')}, {levels.get('S3')} """ # Add strategy flags if present (independent of pivots/fib; keep compact) strategies = market_data.get("strategies") or {} if isinstance(strategies, dict) and strategies: prompt += "\n---\nALGORITHMIC FLAGS / STRATEGY FEATURES\n" per_tf = strategies.get("per_timeframe") or {} mt = strategies.get("multi_timeframe") or {} if isinstance(per_tf, dict): for tf, feats in per_tf.items(): if not feats: continue prompt += f"\nTIMEFRAME: {str(tf).upper()} FLAGS\n{feats}\n" if mt: prompt += f"\nMULTI_TIMEFRAME:\n{mt}\n" # Add Fibonacci retracement / extension levels if present fib_data = market_data.get("fib", {}) if fib_data: prompt += """ --- FIBONACCI LEVELS (retracements and extensions from recent swing) """ for tf, fib in fib_data.items(): if not fib: continue swing_high = fib.get("swing_high") swing_low = fib.get("swing_low") retr = fib.get("retracements", {}) or {} ext = fib.get("extensions", {}) or {} prompt += f""" TIMEFRAME: {tf.upper()} FIBONACCI - Swing High: {swing_high} - Swing Low: {swing_low} - Retracements: {retr} - Extensions: {ext} """ # Add numeric strategy features (computed server-side) if present strategies = market_data.get("strategies", {}) if strategies: per_tf = strategies.get("per_timeframe", {}) or {} mt = strategies.get("multi_timeframe", {}) or {} prompt += """ --- STRATEGY FEATURES (pre-computed, per timeframe) These are numeric/enum flags only; do NOT invent new ones. """ for timeframe, feats in per_tf.items(): trend = feats.get("trend", {}) or {} vol = feats.get("volatility", {}) or {} rb = feats.get("range_breakout", {}) or {} orb = feats.get("opening_range", {}) or {} hod = feats.get("hod_lod", {}) or {} mb = feats.get("momentum_burst", {}) or {} vc = feats.get("volume_climax", {}) or {} pat = feats.get("pattern", {}) or {} fib_zone = feats.get("fib_zone", {}) or {} prompt += f""" TIMEFRAME: {timeframe.upper()} STRATEGY FLAGS - Trend: dir={trend.get('direction')} strength={trend.get('strength')} - Volatility regime: {vol.get('regime')} - Range breakout: status={rb.get('status')} high={rb.get('range_high')} low={rb.get('range_low')} - Opening range: status={orb.get('status')} - HOD/LOD proximity: near_hod={hod.get('near_hod')} near_lod={hod.get('near_lod')} - Momentum burst: status={mb.get('status')} volume_ratio={mb.get('volume_ratio')} - Volume climax: climax={vc.get('climax')} volume_ratio={vc.get('volume_ratio')} - Candle pattern: {pat.get('pattern')} - Fib zone (daily): {fib_zone.get('zone')} """ if mt: prompt += f""" MULTI-TIMEFRAME SUMMARY - Trend alignment: {mt.get('trend_alignment')} """ prompt += """ --- HOW TO THINK ABOUT THIS DATA (GUIDELINES): - Use the candle sequences to approximate: * Trend direction (up / down / sideways) and strength. * Momentum (accelerating, slowing, reversing). * Support and resistance zones. * Volume expansion / exhaustion. * Early breakout or fake breakout attempts. - Use shorter timeframes for fine entry/exit; longer ones for context if available. SCORING & RANKING (VERY IMPORTANT): - Create a numeric "score" from 0 to 100 where: * 0-30 = very weak / avoid trade. * 31-60 = medium quality / only trade with tight risk. * 61-80 = good quality setup. * 81-100 = exceptional / high-conviction setup. - Score MUST be consistent with your BUY/SELL/HOLD decision. Example: HOLD should almost never have score > 40. STOP LOSS AS A ZONE (CRITICAL): - Always think of stop loss as a **zone**, not a single tick. - You MUST output: * A single representative numeric `stop_loss` level, AND * A `stop_loss_zone` object with `upper` and `lower` bounds. - For a BUY decision: * `stop_loss_zone.upper` should be closer to current/entry price. * `stop_loss_zone.lower` should be slightly further away (hard invalidation level). - For a SELL decision (short): * `stop_loss_zone.lower` should be closer to current/entry price. * `stop_loss_zone.upper` should be slightly further away. - "Wick only" breaches into the zone **do NOT** automatically invalidate the idea; design the zone so that only decisive closes beyond the far side of the zone would be true invalidation. EARLY BREAKOUT & ACCUMULATION/DISTRIBUTION (APPROXIMATE): - "early_breakout" should be TRUE only if: * Price is breaking a recent tight range or key level with expanding volume AND strong candles in one direction. - "accumulation_distribution" summary should describe whether price action and volume look like accumulation, distribution or neither. OUTPUT FORMAT (STRICT JSON ONLY): Return ONE valid JSON object with EXACTLY these keys: { "decision": "BUY"|"SELL"|"HOLD", "conviction": "VERY_STRONG"|"HIGH"|"MEDIUM"|"LOW", "trend_label": "", "entry_price": null, "entry_zone": null, "stop_loss": null, "stop_loss_zone": null, "targets": [], "score": 0, "risk_profile": "aggressive"|"conservative", "early_breakout": false, "early_breakout_comment": "", "rationale": ["..."], "technical_indicators": { "trend": "bullish"|"bearish"|"sideways", "momentum": "strong"|"weak"|"neutral", "volume": "high"|"normal"|"low", "support": 0.0, "resistance": 0.0, "accumulation_distribution": "accumulation"|"distribution"|"neutral", "pattern_summary": "" }, "risk_reward_ratio": 0.0, "time_horizon": "intraday", "notes": "" } NON-NEGOTIABLE: - Output ONLY valid JSON. No code blocks. - If decision=HOLD: entry_price=null, entry_zone=null, stop_loss=null, stop_loss_zone=null, targets=[], score<40, conviction=LOW. """ return prompt