quant_engine.py

#!/usr/bin/env python3
"""
QuantEngine - AI-Powered Crypto/Stock Quantitative Analysis & Backtesting Tool

Features:
- Fetch real-time & historical price data (via free APIs)
- Technical indicators: RSI, MACD, Bollinger Bands, EMA, SMA, ATR, OBV
- AI signal generation (rule-based + scoring)
- Strategy backtesting with detailed P&L
- Multi-asset portfolio analysis
- Export signals & reports

Zero paid API keys required - uses free CoinGecko/Yahoo Finance APIs.
"""

import json
import math
import os
import sys
import urllib.request
import urllib.error
from datetime import datetime, timedelta
from collections import defaultdict


# ============================================================
# DATA FETCHING (Free APIs, no keys needed)
# ============================================================

def fetch_crypto_prices(coin_id="bitcoin", vs_currency="usd", days=90):
    """Fetch historical crypto prices from CoinGecko (free, no API key)."""
    url = (
        f"https://api.coingecko.com/api/v3/coins/{coin_id}/market_chart"
        f"?vs_currency={vs_currency}&days={days}&interval=daily"
    )
    try:
        req = urllib.request.Request(url, headers={"User-Agent": "QuantEngine/1.0"})
        with urllib.request.urlopen(req, timeout=15) as resp:
            data = json.loads(resp.read().decode())

        prices = []
        for ts, price in data.get("prices", []):
            dt = datetime.fromtimestamp(ts / 1000).strftime("%Y-%m-%d")
            prices.append({"date": dt, "close": price})

        # Add volume data
        volumes = data.get("total_volumes", [])
        for i, (ts, vol) in enumerate(volumes):
            if i < len(prices):
                prices[i]["volume"] = vol

        print(f"  Fetched {len(prices)} days of {coin_id} price data")
        return prices
    except Exception as e:
        print(f"  Warning: Could not fetch data for {coin_id}: {e}")
        return []


def fetch_multi_crypto(coins=None, vs_currency="usd", days=90):
    """Fetch data for multiple coins."""
    if coins is None:
        coins = ["bitcoin", "ethereum", "solana"]

    all_data = {}
    for coin in coins:
        data = fetch_crypto_prices(coin, vs_currency, days)
        if data:
            all_data[coin] = data
    return all_data


# ============================================================
# TECHNICAL INDICATORS
# ============================================================

def sma(prices, period):
    """Simple Moving Average."""
    result = []
    for i in range(len(prices)):
        if i < period - 1:
            result.append(None)
        else:
            window = [prices[j]["close"] for j in range(i - period + 1, i + 1)]
            result.append(sum(window) / period)
    return result


def ema(prices, period):
    """Exponential Moving Average."""
    result = []
    multiplier = 2 / (period + 1)

    # First EMA = SMA
    if len(prices) < period:
        return [None] * len(prices)

    first_sma = sum(p["close"] for p in prices[:period]) / period
    result = [None] * (period - 1) + [first_sma]

    for i in range(period, len(prices)):
        val = (prices[i]["close"] - result[-1]) * multiplier + result[-1]
        result.append(val)

    return result


def rsi(prices, period=14):
    """Relative Strength Index."""
    if len(prices) < period + 1:
        return [None] * len(prices)

    changes = []
    for i in range(1, len(prices)):
        changes.append(prices[i]["close"] - prices[i-1]["close"])

    result = [None] * period

    # First RSI
    gains = [max(c, 0) for c in changes[:period]]
    losses = [abs(min(c, 0)) for c in changes[:period]]
    avg_gain = sum(gains) / period
    avg_loss = sum(losses) / period

    if avg_loss == 0:
        result.append(100)
    else:
        rs = avg_gain / avg_loss
        result.append(100 - (100 / (1 + rs)))

    # Subsequent RSIs (smoothed)
    for i in range(period, len(changes)):
        gain = max(changes[i], 0)
        loss = abs(min(changes[i], 0))
        avg_gain = (avg_gain * (period - 1) + gain) / period
        avg_loss = (avg_loss * (period - 1) + loss) / period

        if avg_loss == 0:
            result.append(100)
        else:
            rs = avg_gain / avg_loss
            result.append(100 - (100 / (1 + rs)))

    return result


def macd(prices, fast=12, slow=26, signal=9):
    """MACD (Moving Average Convergence Divergence)."""
    ema_fast = ema(prices, fast)
    ema_slow = ema(prices, slow)

    macd_line = []
    for i in range(len(prices)):
        if ema_fast[i] is not None and ema_slow[i] is not None:
            macd_line.append(ema_fast[i] - ema_slow[i])
        else:
            macd_line.append(None)

    # Signal line (EMA of MACD)
    macd_prices = [{"close": v} for v in macd_line if v is not None]
    if len(macd_prices) >= signal:
        signal_line_vals = ema(macd_prices, signal)
        # Pad with None
        pad = len(macd_line) - len(signal_line_vals)
        signal_line = [None] * pad + signal_line_vals
    else:
        signal_line = [None] * len(macd_line)

    # Histogram
    histogram = []
    for i in range(len(macd_line)):
        if macd_line[i] is not None and signal_line[i] is not None:
            histogram.append(macd_line[i] - signal_line[i])
        else:
            histogram.append(None)

    return macd_line, signal_line, histogram


def bollinger_bands(prices, period=20, std_dev=2):
    """Bollinger Bands."""
    middle = sma(prices, period)
    upper = []
    lower = []

    for i in range(len(prices)):
        if middle[i] is None:
            upper.append(None)
            lower.append(None)
        else:
            window = [prices[j]["close"] for j in range(i - period + 1, i + 1)]
            std = math.sqrt(sum((x - middle[i]) ** 2 for x in window) / period)
            upper.append(middle[i] + std_dev * std)
            lower.append(middle[i] - std_dev * std)

    return upper, middle, lower


def atr(prices, period=14):
    """Average True Range (volatility indicator)."""
    if len(prices) < 2:
        return [None] * len(prices)

    true_ranges = [None]
    for i in range(1, len(prices)):
        high = prices[i]["close"] * 1.01  # Approximate high
        low = prices[i]["close"] * 0.99   # Approximate low
        prev_close = prices[i-1]["close"]
        tr = max(high - low, abs(high - prev_close), abs(low - prev_close))
        true_ranges.append(tr)

    result = [None] * period
    if len(true_ranges) > period:
        first_atr = sum(t for t in true_ranges[1:period+1] if t) / period
        result.append(first_atr)

        for i in range(period + 1, len(true_ranges)):
            if true_ranges[i] is not None:
                val = (result[-1] * (period - 1) + true_ranges[i]) / period
                result.append(val)
            else:
                result.append(result[-1])

    return result


# ============================================================
# AI SIGNAL GENERATION
# ============================================================

def generate_signals(prices):
    """
    Generate buy/sell signals based on multiple indicators.
    Returns a score from -100 (strong sell) to +100 (strong buy).
    """
    if len(prices) < 30:
        return [{"date": p["date"], "score": 0, "signal": "HOLD", "reasons": ["Insufficient data"]} for p in prices]

    rsi_vals = rsi(prices)
    macd_line, signal_line, histogram = macd(prices)
    bb_upper, bb_middle, bb_lower = bollinger_bands(prices)
    sma_20 = sma(prices, 20)
    sma_50 = sma(prices, 50)
    ema_12 = ema(prices, 12)

    signals = []
    for i in range(len(prices)):
        score = 0
        reasons = []

        # RSI Signal (-30 to +30)
        if rsi_vals[i] is not None:
            if rsi_vals[i] < 30:
                score += 30
                reasons.append(f"RSI oversold ({rsi_vals[i]:.1f})")
            elif rsi_vals[i] < 40:
                score += 15
                reasons.append(f"RSI low ({rsi_vals[i]:.1f})")
            elif rsi_vals[i] > 70:
                score -= 30
                reasons.append(f"RSI overbought ({rsi_vals[i]:.1f})")
            elif rsi_vals[i] > 60:
                score -= 15
                reasons.append(f"RSI high ({rsi_vals[i]:.1f})")

        # MACD Signal (-25 to +25)
        if histogram[i] is not None:
            if histogram[i] > 0 and (i > 0 and histogram[i-1] is not None and histogram[i-1] <= 0):
                score += 25
                reasons.append("MACD bullish crossover")
            elif histogram[i] < 0 and (i > 0 and histogram[i-1] is not None and histogram[i-1] >= 0):
                score -= 25
                reasons.append("MACD bearish crossover")
            elif histogram[i] > 0:
                score += 10
                reasons.append("MACD positive")
            else:
                score -= 10
                reasons.append("MACD negative")

        # Bollinger Bands Signal (-20 to +20)
        if bb_lower[i] is not None:
            price = prices[i]["close"]
            if price <= bb_lower[i]:
                score += 20
                reasons.append("Price at lower Bollinger Band")
            elif price >= bb_upper[i]:
                score -= 20
                reasons.append("Price at upper Bollinger Band")

        # Moving Average Trend (-15 to +15)
        if sma_20[i] is not None and sma_50[i] is not None:
            if sma_20[i] > sma_50[i]:
                score += 15
                reasons.append("SMA20 > SMA50 (uptrend)")
            else:
                score -= 15
                reasons.append("SMA20 < SMA50 (downtrend)")

        # Price vs EMA (-10 to +10)
        if ema_12[i] is not None:
            if prices[i]["close"] > ema_12[i]:
                score += 10
                reasons.append("Price above EMA12")
            else:
                score -= 10
                reasons.append("Price below EMA12")

        # Clamp score
        score = max(-100, min(100, score))

        # Determine signal
        if score >= 40:
            signal = "STRONG BUY"
        elif score >= 20:
            signal = "BUY"
        elif score <= -40:
            signal = "STRONG SELL"
        elif score <= -20:
            signal = "SELL"
        else:
            signal = "HOLD"

        signals.append({
            "date": prices[i]["date"],
            "price": prices[i]["close"],
            "score": score,
            "signal": signal,
            "reasons": reasons,
            "indicators": {
                "rsi": round(rsi_vals[i], 2) if rsi_vals[i] else None,
                "macd_hist": round(histogram[i], 4) if histogram[i] else None,
                "bb_position": None,
                "sma_20": round(sma_20[i], 2) if sma_20[i] else None,
                "sma_50": round(sma_50[i], 2) if sma_50[i] else None,
            }
        })

    return signals


# ============================================================
# BACKTESTING ENGINE
# ============================================================

def backtest(prices, signals, initial_capital=10000, buy_threshold=20, sell_threshold=-20):
    """
    Backtest a signal-based strategy.
    Buy when score >= buy_threshold, sell when score <= sell_threshold.
    """
    capital = initial_capital
    position = 0  # Number of units held
    entry_price = 0
    trades = []
    equity_curve = []

    for i in range(len(prices)):
        price = prices[i]["close"]
        score = signals[i]["score"]

        # Buy signal
        if score >= buy_threshold and position == 0:
            position = capital / price
            entry_price = price
            capital = 0
            trades.append({
                "type": "BUY",
                "date": prices[i]["date"],
                "price": price,
                "score": score,
                "units": position,
            })

        # Sell signal
        elif score <= sell_threshold and position > 0:
            capital = position * price
            pnl = capital - (position * entry_price)
            pnl_pct = (price - entry_price) / entry_price * 100
            trades.append({
                "type": "SELL",
                "date": prices[i]["date"],
                "price": price,
                "score": score,
                "units": position,
                "pnl": round(pnl, 2),
                "pnl_pct": round(pnl_pct, 2),
            })
            position = 0
            entry_price = 0

        # Track equity
        current_equity = capital + (position * price)
        equity_curve.append({
            "date": prices[i]["date"],
            "equity": round(current_equity, 2),
            "price": price,
        })

    # Final equity
    final_equity = capital + (position * prices[-1]["close"] if position > 0 else 0)
    total_return = (final_equity - initial_capital) / initial_capital * 100

    # Calculate stats
    winning_trades = [t for t in trades if t["type"] == "SELL" and t.get("pnl", 0) > 0]
    losing_trades = [t for t in trades if t["type"] == "SELL" and t.get("pnl", 0) <= 0]
    sell_trades = [t for t in trades if t["type"] == "SELL"]

    max_equity = max(e["equity"] for e in equity_curve) if equity_curve else initial_capital
    min_equity_after_max = initial_capital
    max_drawdown = 0
    peak = initial_capital
    for e in equity_curve:
        if e["equity"] > peak:
            peak = e["equity"]
        dd = (peak - e["equity"]) / peak * 100
        if dd > max_drawdown:
            max_drawdown = dd

    results = {
        "initial_capital": initial_capital,
        "final_equity": round(final_equity, 2),
        "total_return_pct": round(total_return, 2),
        "total_trades": len(sell_trades),
        "winning_trades": len(winning_trades),
        "losing_trades": len(losing_trades),
        "win_rate": round(len(winning_trades) / len(sell_trades) * 100, 1) if sell_trades else 0,
        "max_drawdown_pct": round(max_drawdown, 2),
        "avg_profit": round(sum(t.get("pnl", 0) for t in winning_trades) / len(winning_trades), 2) if winning_trades else 0,
        "avg_loss": round(sum(t.get("pnl", 0) for t in losing_trades) / len(losing_trades), 2) if losing_trades else 0,
        "trades": trades,
        "equity_curve": equity_curve,
    }

    return results


# ============================================================
# REPORTING
# ============================================================

def print_signal_report(signals, last_n=10):
    """Print recent signals."""
    print("\n" + "=" * 70)
    print("  TRADING SIGNALS (Last {} days)".format(last_n))
    print("=" * 70)
    print(f"{'Date':<12} {'Price':>10} {'Score':>6} {'Signal':<12} {'Key Reason'}")
    print("-" * 70)

    for s in signals[-last_n:]:
        reason = s["reasons"][0] if s["reasons"] else ""
        print(f"{s['date']:<12} {s['price']:>10.2f} {s['score']:>6d} {s['signal']:<12} {reason}")


def print_backtest_report(results):
    """Print backtest results."""
    print("\n" + "=" * 70)
    print("  BACKTEST RESULTS")
    print("=" * 70)
    print(f"  Initial Capital:   ${results['initial_capital']:,.2f}")
    print(f"  Final Equity:      ${results['final_equity']:,.2f}")
    print(f"  Total Return:      {results['total_return_pct']:+.2f}%")
    print(f"  Max Drawdown:      {results['max_drawdown_pct']:.2f}%")
    print(f"  Total Trades:      {results['total_trades']}")
    print(f"  Win Rate:          {results['win_rate']}%")
    print(f"  Avg Profit/Trade:  ${results['avg_profit']:,.2f}")
    print(f"  Avg Loss/Trade:    ${results['avg_loss']:,.2f}")
    print("-" * 70)

    if results["trades"]:
        print("\n  Trade History:")
        print(f"  {'Type':<6} {'Date':<12} {'Price':>10} {'P&L':>10} {'P&L%':>8}")
        print("  " + "-" * 48)
        for t in results["trades"]:
            pnl = f"${t.get('pnl', 0):,.2f}" if "pnl" in t else ""
            pnl_pct = f"{t.get('pnl_pct', 0):+.2f}%" if "pnl_pct" in t else ""
            print(f"  {t['type']:<6} {t['date']:<12} {t['price']:>10.2f} {pnl:>10} {pnl_pct:>8}")


def export_signals_json(signals, filepath):
    """Export signals to JSON."""
    with open(filepath, "w") as f:
        json.dump(signals, f, indent=2)
    print(f"  Signals exported to {filepath}")


def export_backtest_json(results, filepath):
    """Export backtest results to JSON."""
    with open(filepath, "w") as f:
        json.dump(results, f, indent=2)
    print(f"  Backtest results exported to {filepath}")


# ============================================================
# MAIN
# ============================================================

def run_analysis(coin="bitcoin", days=90, capital=10000):
    """Run full analysis pipeline for a coin."""
    print(f"\n{'='*70}")
    print(f"  QuantEngine Analysis: {coin.upper()}")
    print(f"  Period: {days} days | Capital: ${capital:,.2f}")
    print(f"{'='*70}\n")

    # Fetch data
    print("[1/4] Fetching price data...")
    prices = fetch_crypto_prices(coin, days=days)
    if not prices:
        print("  ERROR: Could not fetch data. Check internet connection.")
        return None

    # Generate signals
    print("[2/4] Computing technical indicators & signals...")
    signals = generate_signals(prices)

    # Print signals
    print("[3/4] Signal report:")
    print_signal_report(signals, last_n=15)

    # Backtest
    print("\n[4/4] Running backtest...")
    results = backtest(prices, signals, initial_capital=capital)
    print_backtest_report(results)

    # Current recommendation
    latest = signals[-1]
    print(f"\n{'='*70}")
    print(f"  CURRENT RECOMMENDATION: {latest['signal']} (score: {latest['score']})")
    print(f"  Price: ${latest['price']:,.2f}")
    print(f"  Reasons:")
    for r in latest["reasons"]:
        print(f"    - {r}")
    print(f"{'='*70}\n")

    return {
        "prices": prices,
        "signals": signals,
        "backtest": results,
    }


def run_portfolio_analysis(coins=None, days=90, capital=10000):
    """Run analysis on multiple coins and rank them."""
    if coins is None:
        coins = ["bitcoin", "ethereum", "solana", "cardano", "polkadot"]

    print(f"\n{'='*70}")
    print(f"  PORTFOLIO SCANNER - {len(coins)} assets")
    print(f"{'='*70}\n")

    results = []
    for coin in coins:
        prices = fetch_crypto_prices(coin, days=days)
        if not prices:
            continue
        signals = generate_signals(prices)
        bt = backtest(prices, signals, initial_capital=capital)
        latest = signals[-1] if signals else None

        results.append({
            "coin": coin,
            "price": prices[-1]["close"] if prices else 0,
            "signal": latest["signal"] if latest else "N/A",
            "score": latest["score"] if latest else 0,
            "backtest_return": bt["total_return_pct"],
            "win_rate": bt["win_rate"],
            "max_drawdown": bt["max_drawdown_pct"],
        })

    # Sort by score
    results.sort(key=lambda x: x["score"], reverse=True)

    print(f"\n{'='*70}")
    print(f"  PORTFOLIO RANKING")
    print(f"{'='*70}")
    print(f"{'Rank':<5} {'Coin':<12} {'Price':>10} {'Signal':<12} {'Score':>6} {'BT Return':>10} {'Win%':>6}")
    print("-" * 70)
    for i, r in enumerate(results, 1):
        print(f"{i:<5} {r['coin']:<12} ${r['price']:>9.2f} {r['signal']:<12} {r['score']:>6} {r['backtest_return']:>9.2f}% {r['win_rate']:>5.1f}%")

    print(f"\n  Top pick: {results[0]['coin'].upper()} ({results[0]['signal']}, score {results[0]['score']})")

    return results


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="QuantEngine - AI Quantitative Trading Analysis")
    parser.add_argument("command", choices=["analyze", "scan", "signals"],
                       help="analyze: single coin | scan: portfolio scanner | signals: export signals")
    parser.add_argument("--coin", default="bitcoin", help="Coin ID (default: bitcoin)")
    parser.add_argument("--coins", nargs="+", default=None, help="Multiple coin IDs for scan")
    parser.add_argument("--days", type=int, default=90, help="Days of history (default: 90)")
    parser.add_argument("--capital", type=float, default=10000, help="Starting capital (default: 10000)")
    parser.add_argument("--output", "-o", help="Export results to JSON file")

    args = parser.parse_args()

    if args.command == "analyze":
        result = run_analysis(args.coin, args.days, args.capital)
        if result and args.output:
            export_backtest_json(result["backtest"], args.output)

    elif args.command == "scan":
        coins = args.coins or ["bitcoin", "ethereum", "solana", "cardano", "polkadot"]
        results = run_portfolio_analysis(coins, args.days, args.capital)
        if args.output:
            with open(args.output, "w") as f:
                json.dump(results, f, indent=2)
            print(f"  Results exported to {args.output}")

    elif args.command == "signals":
        prices = fetch_crypto_prices(args.coin, days=args.days)
        if prices:
            signals = generate_signals(prices)
            print_signal_report(signals, last_n=30)
            if args.output:
                export_signals_json(signals, args.output)