Signal Alerts

# ta-config.yaml — alert settings

watchlist:
  tickers:
    - AAPL
    - MSFT
    - NVDA
    - TSLA
    - SPY

signals:
  rsi_oversold:    30     # Send alert when RSI drops below this
  rsi_overbought:  70     # Send alert when RSI rises above this
  golden_cross:    true   # Alert when 50-day SMA crosses above 200-day
  death_cross:     true   # Alert when 50-day SMA crosses below 200-day
  macd_cross:      true   # Alert on MACD line/signal crossover
  bb_squeeze:      true   # Alert when Bollinger Bands compress

alerts:
  email: "you@example.com"       # Where to send alerts
  cooldown_hours: 24             # Don't re-alert the same signal for 24 hours
  daily_digest: true             # true = one daily summary | false = individual alerts
  send_time: "16:30"             # When to send the digest (24h format, your timezone)

  # Email server settings (use your own SMTP or a service like SendGrid)
  smtp_host: "localhost"         # or "smtp.gmail.com" for Gmail
  smtp_port: 587

# ta_alerts.py — OpenClaw Signal Alert System
# Monitors your watchlist and emails you when signal conditions trigger

import yaml
import yfinance as yf
import pandas_ta as ta
import smtplib
import time
import json
import os
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from datetime import datetime, timedelta

# ── Load your config ──────────────────────────────────────────────
def load_config(path: str = "ta-config.yaml") -> dict:
    """Load configuration from YAML file."""
    with open(path) as f:
        return yaml.safe_load(f)

cfg     = load_config()
TICKERS = cfg["watchlist"]["tickers"]
SIGNALS = cfg["signals"]
ALERTS  = cfg["alerts"]

print(f"🦞 OpenClaw Alerts | Watching {len(TICKERS)} tickers")
print(f"   Mode: {'Daily Digest' if ALERTS.get('daily_digest') else 'Real-Time'}")
print(f"   Cooldown: {ALERTS['cooldown_hours']} hours")

# ── Cooldown tracker — prevents duplicate alerts ──────────────────
# Stored in a local file so it survives script restarts
COOLDOWN_FILE = ".alert_cooldowns.json"

def load_cooldowns() -> dict:
    """Load the cooldown record from disk."""
    if os.path.exists(COOLDOWN_FILE):
        with open(COOLDOWN_FILE) as f:
            return json.load(f)
    return {}

def save_cooldowns(cooldowns: dict) -> None:
    """Save cooldown record to disk."""
    with open(COOLDOWN_FILE, "w") as f:
        json.dump(cooldowns, f)

def is_on_cooldown(ticker: str, signal: str, cooldown_hours: int) -> bool:
    """
    Returns True if this ticker+signal was already alerted recently.
    Prevents the same signal from alerting multiple days in a row.
    """
    cooldowns = load_cooldowns()
    key = f"{ticker}_{signal}"
    if key not in cooldowns:
        return False
    last_alert = datetime.fromisoformat(cooldowns[key])
    return datetime.now() - last_alert < timedelta(hours=cooldown_hours)

def mark_alerted(ticker: str, signal: str) -> None:
    """Record the current time as the last alert for this ticker+signal."""
    cooldowns = load_cooldowns()
    cooldowns[f"{ticker}_{signal}"] = datetime.now().isoformat()
    save_cooldowns(cooldowns)

# ── Check one ticker for all signal conditions ────────────────────
def check_ticker(ticker: str) -> list:
    """
    Returns a list of triggered signals for this ticker.
    An empty list means nothing fired — no alert needed.
    """
    try:
        # Download 1 year of daily data
        df = yf.download(ticker, period="1y", interval="1d", progress=False)
        if df.empty or len(df) < 210:
            return []

        # Normalize column names
        df.columns = [c.lower() for c in df.columns]
        df = df[["open","high","low","close","volume"]].dropna()

        # Calculate technical indicators
        df["sma50"]  = ta.sma(df["close"], 50)
        df["sma200"] = ta.sma(df["close"], 200)
        df["rsi"]    = ta.rsi(df["close"], 14)

        macd = ta.macd(df["close"], 12, 26, 9)
        df["macd"]     = macd[[c for c in macd.columns if c.startswith("MACD_")][0]]
        df["macd_sig"] = macd[[c for c in macd.columns if c.startswith("MACDs_")][0]]

        bb = ta.bbands(df["close"], 20, 2)
        df["bb_bw"] = bb[[c for c in bb.columns if "BBB" in c][0]]
        df = df.dropna()

        if len(df) < 2:
            return []

        # Get current and previous bar
        cur  = df.iloc[-1]
        prev = df.iloc[-2]
        fired = []

        # ── RSI checks
        if SIGNALS.get("rsi_oversold") and cur["rsi"] < SIGNALS["rsi_oversold"]:
            fired.append(("RSI_OVERSOLD",
                f"RSI hit {cur['rsi']:.1f} — below your oversold threshold of {SIGNALS['rsi_oversold']}"))

        if SIGNALS.get("rsi_overbought") and cur["rsi"] > SIGNALS["rsi_overbought"]:
            fired.append(("RSI_OVERBOUGHT",
                f"RSI hit {cur['rsi']:.1f} — above your overbought threshold of {SIGNALS['rsi_overbought']}"))

        # ── Moving Average Crosses
        if SIGNALS.get("golden_cross"):
            if cur["sma50"] > cur["sma200"] and prev["sma50"] <= prev["sma200"]:
                fired.append(("GOLDEN_CROSS",
                    f"50-day SMA ({cur['sma50']:.2f}) just crossed above 200-day SMA ({cur['sma200']:.2f})"))

        if SIGNALS.get("death_cross"):
            if cur["sma50"] < cur["sma200"] and prev["sma50"] >= prev["sma200"]:
                fired.append(("DEATH_CROSS",
                    f"50-day SMA ({cur['sma50']:.2f}) just crossed below 200-day SMA ({cur['sma200']:.2f})"))

        # ── MACD Crossover
        if SIGNALS.get("macd_cross"):
            if cur["macd"] > cur["macd_sig"] and prev["macd"] <= prev["macd_sig"]:
                fired.append(("MACD_BULL_CROSS",
                    f"MACD line crossed above Signal line — momentum turning bullish"))

            if cur["macd"] < cur["macd_sig"] and prev["macd"] >= prev["macd_sig"]:
                fired.append(("MACD_BEAR_CROSS",
                    f"MACD line crossed below Signal line — momentum turning bearish"))

        # ── Bollinger Squeeze
        if SIGNALS.get("bb_squeeze"):
            if cur["bb_bw"] < df["bb_bw"].quantile(0.20):
                fired.append(("BB_SQUEEZE",
                    f"Bollinger bandwidth at historic low — volatility compression, watch for breakout"))

        # ── Filter out signals on cooldown
        filtered = []
        for signal_name, description in fired:
            if not is_on_cooldown(ticker, signal_name, ALERTS["cooldown_hours"]):
                filtered.append((signal_name, description, cur["close"]))
                mark_alerted(ticker, signal_name)

        return filtered

    except Exception as e:
        # Silently skip tickers with errors (delisted, no data, etc.)
        return []

# ── Send an email alert ───────────────────────────────────────────
def send_email(subject: str, body: str) -> None:
    """Send an email using SMTP settings from ta-config.yaml."""
    msg = MIMEMultipart()
    msg["Subject"] = subject
    msg["From"]    = f"OpenClaw Alerts <{ALERTS.get('smtp_from', 'openclaw@localhost')}>"
    msg["To"]      = ALERTS["email"]
    msg.attach(MIMEText(body, "plain"))

    # Connect to SMTP server and send
    with smtplib.SMTP(ALERTS.get("smtp_host", "localhost"),
                      ALERTS.get("smtp_port", 587)) as s:
        s.ehlo()
        if ALERTS.get("smtp_tls", False):
            s.starttls()
        if ALERTS.get("smtp_user"):
            s.login(ALERTS["smtp_user"], ALERTS["smtp_password"])
        s.send_message(msg)

    print(f"✓ Email sent: {subject}")

# ── Main run — check all tickers ─────────────────────────────────
all_signals = []

for ticker in TICKERS:
    signals = check_ticker(ticker)
    if signals:
        for sig_name, description, price in signals:
            print(f"  🚨 {ticker}: {sig_name}")
            all_signals.append({
                "ticker": ticker, "signal": sig_name,
                "description": description, "price": price,
                "time": datetime.now().strftime("%H:%M")
            })

            # Real-time mode: send immediately
            if not ALERTS.get("daily_digest"):
                send_email(
                    subject=f"[OpenClaw] {ticker}: {sig_name}",
                    body=(
                        f"🦞 OpenClaw TA Alert\n"
                        f"{'─'*40}\n"
                        f"Ticker:  {ticker}\n"
                        f"Signal:  {sig_name}\n"
                        f"Price:   ${price:,.2f}\n"
                        f"Detail:  {description}\n"
                        f"Time:    {datetime.now().strftime('%Y-%m-%d %H:%M')}\n"
                        f"{'─'*40}\n"
                        f"Not financial advice. | learnopenclaw.org"
                    )
                )
    time.sleep(0.5)  # Rate limit to avoid overwhelming API

# ── Daily digest mode: send one summary ──────────────────────────
if ALERTS.get("daily_digest") and all_signals:
    lines = [
        f"🦞 OpenClaw TA Daily Digest — {datetime.now().strftime('%Y-%m-%d')}",
        f"{'─'*45}",
        f"{len(all_signals)} signal(s) across your watchlist:\n"
    ]

    for sig in all_signals:
        lines.append(f"  {sig['ticker']:<8} {sig['signal']:<20} @ ${sig['price']:,.2f}")
        lines.append(f"           {sig['description']}\n")

    lines += [
        f"{'─'*45}",
        "Not financial advice. | learnopenclaw.org"
    ]

    send_email(
        subject=f"[OpenClaw] Daily TA Digest — {len(all_signals)} signals — {datetime.now().strftime('%b %d')}",
        body="\n".join(lines)
    )
elif not all_signals:
    print("✓ No signals today — no email sent")

# Run every day at 4:15pm Mon–Fri (after US market close)
# Add this line to your crontab: crontab -e
15 16 * * 1-5 cd /path/to/openclaw && python3 ta_alerts.py >> alerts.log 2>&1