AI-Powered Stock Watchlist Analysis with Daily Alerts and PostgreSQL Logging

id: ai-stock-analysis-alerts
namespace: company.team

inputs:
  - id: watchlist
    type: ARRAY
    itemType: STRING
    displayName: Stock watchlist (tickers)
    description: List of stock ticker symbols to analyze
    defaults:
      - "AAPL"
      - "MSFT"
      - "GOOGL"
      - "NVDA"
      - "TSLA"
  - id: analysis_horizon
    type: STRING
    displayName: Analysis horizon
    description: Time horizon for AI recommendations (e.g., "short-term (1-2
      weeks)", "medium-term (1-3 months)")
    defaults: "short-term (1-2 weeks)"
  - id: risk_profile
    type: SELECT
    displayName: Risk profile
    description: Investor risk tolerance, shapes the AI recommendation style
    values:
      - conservative
      - moderate
      - aggressive
    defaults: "moderate"

tasks:
  - id: fetch_market_data
    type: io.kestra.plugin.core.flow.ForEach
    values: "{{ inputs.watchlist }}"
    concurrencyLimit: 5
    tasks:
      - id: fetch_quote
        type: io.kestra.plugin.core.http.Request
        uri: "https://query1.finance.yahoo.com/v8/finance/chart/{{ taskrun.value
          }}?interval=1d&range=30d"
        method: GET
        headers:
          User-Agent: "Mozilla/5.0"

  - id: prepare_analysis_data
    type: io.kestra.plugin.scripts.python.Script
    inputFiles:
      quotes_raw.json: "{{ outputs.fetch_market_data | json }}"
    script: |
      import json
      from kestra import Kestra

      with open("quotes_raw.json") as f:
          raw = json.load(f)

      stocks = []
      for ticker, task_output in raw.items():
          body = task_output.get("body", "{}")
          if isinstance(body, str):
              try:
                  data = json.loads(body)
              except Exception:
                  continue
          else:
              data = body

          try:
              result = data["chart"]["result"][0]
              meta = result["meta"]
              closes = result["indicators"]["quote"][0].get("closes") or result["indicators"]["quote"][0].get("close", [])
              closes = [c for c in closes if c is not None]

              if len(closes) < 2:
                  continue

              current_price = closes[-1]
              prev_price = closes[-2]
              price_30d_ago = closes[0]

              day_change_pct = ((current_price - prev_price) / prev_price) * 100
              month_change_pct = ((current_price - price_30d_ago) / price_30d_ago) * 100

              # Simple RSI approximation (14-day)
              gains = [max(0, closes[i] - closes[i-1]) for i in range(1, len(closes))]
              losses = [max(0, closes[i-1] - closes[i]) for i in range(1, len(closes))]
              avg_gain = sum(gains[-14:]) / 14 if len(gains) >= 14 else sum(gains) / len(gains) if gains else 0
              avg_loss = sum(losses[-14:]) / 14 if len(losses) >= 14 else sum(losses) / len(losses) if losses else 0
              rsi = 100 - (100 / (1 + avg_gain / avg_loss)) if avg_loss > 0 else 100

              stocks.append({
                  "ticker": meta.get("symbol", ticker),
                  "current_price": round(current_price, 2),
                  "day_change_pct": round(day_change_pct, 2),
                  "month_change_pct": round(month_change_pct, 2),
                  "rsi_14d": round(rsi, 1),
                  "52w_high": meta.get("fiftyTwoWeekHigh"),
                  "52w_low": meta.get("fiftyTwoWeekLow"),
                  "currency": meta.get("currency", "USD"),
                  "price_history_30d": [round(c, 2) for c in closes[-10:]]
              })
          except (KeyError, IndexError, ZeroDivisionError, TypeError):
              continue

      stocks.sort(key=lambda x: abs(x["day_change_pct"]), reverse=True)
      Kestra.outputs({"stocks": stocks, "count": len(stocks)})

  - id: generate_ai_analysis
    type: io.kestra.plugin.openai.ChatCompletion
    apiKey: "{{ secret('OPENAI_API_KEY') }}"
    model: gpt-4o
    maxTokens: 1500
    jsonResponseSchema: |
      {
        "type": "object",
        "properties": {
          "market_summary": {"type": "string"},
          "recommendations": {
            "type": "array",
            "items": {
              "type": "object",
              "properties": {
                "ticker": {"type": "string"},
                "action": {"type": "string", "enum": ["BUY", "HOLD", "SELL", "WATCH"]},
                "confidence": {"type": "number", "minimum": 0, "maximum": 1},
                "rationale": {"type": "string"},
                "key_levels": {"type": "string"}
              },
              "required": ["ticker", "action", "confidence", "rationale"]
            }
          },
          "risk_flags": {"type": "array", "items": {"type": "string"}}
        },
        "required": ["market_summary", "recommendations"]
      }
    messages:
      - role: system
        content: |
          You are a quantitative stock analyst providing technical analysis recommendations.
          Risk profile: {{ inputs.risk_profile }}
          Time horizon: {{ inputs.analysis_horizon }}

          For each stock, analyze: momentum (day/month % change), RSI (overbought >70, oversold <30),
          proximity to 52-week highs/lows, and relative performance.

          Provide actionable recommendations (BUY/HOLD/SELL/WATCH) calibrated to the risk profile:
          - conservative: prioritize capital preservation, flag high volatility
          - moderate: balanced risk/reward, standard momentum signals
          - aggressive: favor momentum plays, higher confidence thresholds

          This is for educational/informational purposes only, not financial advice.
          Return structured JSON only.
      - role: user
        content: |
          Analyze these stocks for {{ inputs.analysis_horizon }} {{ inputs.risk_profile }} strategy:
          {{ outputs.prepare_analysis_data.vars.stocks | json }}

  - id: log_to_postgres
    type: io.kestra.plugin.jdbc.postgresql.Queries
    fetchType: NONE
    sql: |
      CREATE TABLE IF NOT EXISTS stock_analysis_log (
          id SERIAL PRIMARY KEY,
          ticker TEXT,
          action TEXT,
          confidence NUMERIC,
          current_price NUMERIC,
          day_change_pct NUMERIC,
          rsi_14d NUMERIC,
          rationale TEXT,
          risk_profile TEXT,
          analyzed_at TIMESTAMP DEFAULT NOW()
      );

      INSERT INTO stock_analysis_log
          (ticker, action, confidence, current_price, day_change_pct, rsi_14d, rationale, risk_profile)
      SELECT
          r->>'ticker',
          r->>'action',
          (r->>'confidence')::numeric,
          (s->>'current_price')::numeric,
          (s->>'day_change_pct')::numeric,
          (s->>'rsi_14d')::numeric,
          r->>'rationale',
          '{{ inputs.risk_profile }}'
      FROM json_array_elements('{{ outputs.generate_ai_analysis.choices[0].message.content | json | jq(".recommendations") | first | json }}'::json) AS r
      JOIN json_array_elements('{{ outputs.prepare_analysis_data.vars.stocks | json }}'::json) AS s
        ON r->>'ticker' = s->>'ticker';

  - id: send_alerts
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: |
      :chart_with_upwards_trend: *Daily Stock Analysis, {{ now() | date('yyyy-MM-dd') }}*
      _{{ inputs.risk_profile | title }} profile · {{ inputs.analysis_horizon }}_

      *Market Summary:*
      {{ outputs.generate_ai_analysis.choices[0].message.content | json | jq(".market_summary") | first }}

      *Recommendations ({{ outputs.prepare_analysis_data.vars.count }} stocks analyzed):*
      {{ outputs.generate_ai_analysis.choices[0].message.content | json | jq('[.recommendations[] | "• " + .ticker + " → " + .action + " (" + (.confidence * 100 | floor | tostring) + "% conf): " + .rationale] | join("\n")') | first }}

      _⚠️ For informational purposes only, not financial advice._

pluginDefaults:
  - type: io.kestra.plugin.jdbc.postgresql.Queries
    values:
      url: "{{ secret('POSTGRES_URL') }}"
      username: "{{ secret('POSTGRES_USER') }}"
      password: "{{ secret('POSTGRES_PASSWORD') }}"

triggers:
  - id: market_open_daily
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 14 * * 1-5"
    description: Run Monday–Friday at 14:00 UTC (09:00 ET market open) for daily
      pre-market analysis