How to Build an AI Market Analyst That Runs 24/7

import requests
import json
from datetime import datetime
from pathlib import Path

API_KEY = "your_api_key_here"
BASE_URL = "https://tickatlas.com/v1"
HEADERS = {"X-API-Key": API_KEY}

WATCHLIST = ["EURUSD", "GBPUSD", "USDJPY", "XAUUSD", "BTCUSD"]
INDICATORS = "RSI_14,MACD,BBANDS_20,ADX_14,EMA_9,EMA_21,ATR_14"

def collect_snapshot() -> dict:
    """Fetch current indicators for all watchlist symbols."""
    snapshot = {
        "timestamp": datetime.utcnow().isoformat(),
        "symbols": {}
    }

    for symbol in WATCHLIST:
        resp = requests.get(f"{BASE_URL}/indicators", params={
            "symbol": symbol,
            "timeframe": "H1",
            "indicators": INDICATORS
        }, headers=HEADERS)

        if resp.status_code == 200:
            data = resp.json()["data"]
            snapshot["symbols"][symbol] = {
                "indicators": data["indicators"],
                "price": data.get("price", {}),
            }

    return snapshot

def save_snapshot(snapshot: dict):
    """Persist snapshot for historical comparison."""
    path = Path("snapshots")
    path.mkdir(exist_ok=True)
    filename = f"snapshot_{snapshot['timestamp'][:10]}.jsonl"
    with open(path / filename, "a") as f:
        f.write(json.dumps(snapshot) + "\n")

def detect_alerts(current: dict, previous: dict) -> list[dict]:
    """Compare snapshots and detect significant changes."""
    alerts = []

    for symbol in WATCHLIST:
        curr = current["symbols"].get(symbol, {}).get("indicators", {})
        prev = previous["symbols"].get(symbol, {}).get("indicators", {})
        if not curr or not prev:
            continue

        rsi = curr.get("RSI_14", {}).get("value", 50)
        prev_rsi = prev.get("RSI_14", {}).get("value", 50)

        # RSI entering oversold
        if rsi < 30 and prev_rsi >= 30:
            alerts.append({
                "symbol": symbol,
                "type": "RSI_OVERSOLD",
                "value": rsi,
                "message": f"{symbol} RSI entered oversold territory ({rsi:.1f})"
            })

        # RSI entering overbought
        if rsi > 70 and prev_rsi <= 70:
            alerts.append({
                "symbol": symbol,
                "type": "RSI_OVERBOUGHT",
                "value": rsi,
                "message": f"{symbol} RSI entered overbought territory ({rsi:.1f})"
            })

        # MACD histogram crossover
        macd_hist = curr.get("MACD", {}).get("histogram", 0)
        prev_macd_hist = prev.get("MACD", {}).get("histogram", 0)

        if macd_hist > 0 and prev_macd_hist <= 0:
            alerts.append({
                "symbol": symbol,
                "type": "MACD_BULLISH_CROSS",
                "message": f"{symbol} MACD histogram turned bullish"
            })

    return alerts

import anthropic

llm = anthropic.Anthropic()

def generate_analysis(snapshot: dict,
                      alerts: list[dict] = None) -> str:
    """Generate AI market analysis from snapshot data."""
    prompt = f"""Generate a concise market analysis briefing based on
this indicator data:

{json.dumps(snapshot['symbols'], indent=2)}

{"Active alerts: " + json.dumps(alerts) if alerts else "No active alerts."}

Format as a professional market briefing:
1. Market Overview (2-3 sentences)
2. Per-symbol analysis (key findings only, skip symbols with no signal)
3. Watchlist items (what to monitor in the next few hours)

Be specific with numbers. Mention RSI levels, MACD direction,
Bollinger Band position, and ADX trend strength."""

    response = llm.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1500,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text

TELEGRAM_BOT_TOKEN = "your_bot_token"
TELEGRAM_CHAT_ID = "your_chat_id"

def send_telegram(message: str):
    """Send analysis to Telegram."""
    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    requests.post(url, json={
        "chat_id": TELEGRAM_CHAT_ID,
        "text": message,
        "parse_mode": "Markdown"
    })

import time
import schedule

previous_snapshot = None

def market_check():
    """Runs every 15 minutes."""
    global previous_snapshot
    current = collect_snapshot()
    save_snapshot(current)

    if previous_snapshot:
        alerts = detect_alerts(current, previous_snapshot)
        if alerts:
            analysis = generate_analysis(current, alerts)
            send_telegram(f"*Alert*\n\n{analysis}")

    previous_snapshot = current

def daily_briefing():
    """Runs once at 7:00 AM UTC."""
    snapshot = collect_snapshot()
    analysis = generate_analysis(snapshot)
    send_telegram(f"*Daily Market Briefing*\n\n{analysis}")

# Schedule
schedule.every(15).minutes.do(market_check)
schedule.every().day.at("07:00").do(daily_briefing)

print("AI Market Analyst started. Running 24/7...")
while True:
    schedule.run_pending()
    time.sleep(60)