app.py

# ============================================
# АВТО-УСТАНОВКА ПАКЕТОВ
# ============================================
import subprocess, sys, importlib

REQUIRED_PACKAGES = {
    'numpy': 'numpy',
    'pandas': 'pandas',
    'httpx': 'httpx',
    'scipy': 'scipy',
    'fastapi': 'fastapi',
    'uvicorn': 'uvicorn',
    'requests': 'requests'
}

for module_name, pip_name in REQUIRED_PACKAGES.items():
    try:
        importlib.import_module(module_name)
    except ImportError:
        print(f"📦 Устанавливаю {pip_name}...")
        subprocess.check_call([sys.executable, "-m", "pip", "install", pip_name])
        print(f"✅ {pip_name} установлен!")

# ============================================
# 👑 TOMIRIS SPACE 25 v4.0 — INTERMARKET CORRELATION ENGINE (УСИЛЕННЫЙ)
# ============================================
import os, time, json, logging, asyncio
from typing import Dict, Any, List, Optional, Tuple
from datetime import datetime, timezone
from collections import deque
import numpy as np
import pandas as pd
from scipy import stats
import httpx
from fastapi import FastAPI, Query

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
logger = logging.getLogger("Space25_Correlation")

# ================= КОНФИГУРАЦИЯ =================
SPACE_ID = 25
SPACE_NAME = "Correlation Engine"

SYMBOLS = ["XAU/USD", "ETH/USD", "SOL/USD"]

CORRELATION_PAIRS = {
    "XAU/USD": ["DXY", "TIPS", "SPY", "QQQ", "GLD", "BTC/USD"],
    "ETH/USD": ["BTC/USD", "XAU/USD", "SPY", "QQQ", "SOL/USD", "DXY"],
    "SOL/USD": ["BTC/USD", "ETH/USD", "QQQ", "SPY"]
}

TICKER_MAP = {
    "XAU/USD": "XAU/USD", "ETH/USD": "ETH/USD", "SOL/USD": "SOL/USD",
    "BTC/USD": "BTC/USD", "DXY": "DXY", "SPY": "SPY", "QQQ": "QQQ",
    "TIPS": "TIP", "GLD": "GLD"
}

HUB_URL = "https://TOMI-HUB-HUB-FINAL.hf.space"
HUB_SECRET = os.getenv("HUB_SECRET", "TomyrisUltraSecret2026!")
TWELVE_DATA_KEY = os.getenv("TWELVE_DATA_KEY", "")

STARTUP_SLEEP = int(os.getenv("STARTUP_SLEEP", "120"))
AUTO_SEND_INTERVAL = int(os.getenv("AUTO_SEND_INTERVAL", "300"))

logger.info(f"🔗 Хаб: {HUB_URL} | Старт: {STARTUP_SLEEP}с | Интервал: {AUTO_SEND_INTERVAL}с")

HISTORY_FILE = "correlation_history.json"
CACHE_TTL = 300

# ================= HTTP КЛИЕНТ =================
http_client = httpx.AsyncClient(timeout=15.0)

def hub_headers():
    return {"X-Hub-Secret": HUB_SECRET, "Content-Type": "application/json"}

async def log_to_hub(event_type: str, message: str, details: dict = None):
    try:
        await http_client.post(
            f"{HUB_URL}/log",
            json={"space_id": str(SPACE_ID), "event_type": event_type, "message": message, "details": details or {}},
            headers=hub_headers(), timeout=5
        )
    except: pass

# ================= ИСТОРИЯ ДЛЯ Z-SCORE =================
SCORE_HISTORY = {sym: deque(maxlen=200) for sym in SYMBOLS}
CORR_HISTORY = {}  # Для каждой пары

def calculate_zscore(current: float, history: deque) -> float:
    if len(history) < 10: return 0.0
    arr = np.array(list(history))
    mean, std = arr.mean(), arr.std()
    if std == 0: return 0.0
    return (current - mean) / std

# ================= ГЛОБАЛЬНЫЙ КЭШ И BREAKER =================
price_cache = {}
cache_timestamps = {}
CIRCUIT_BREAKER = {}

def breaker_open(name: str) -> bool:
    info = CIRCUIT_BREAKER.get(name)
    if not info: return False
    if info["fails"] < 3: return False
    if time.time() - info["last_fail"] > 600:
        CIRCUIT_BREAKER[name] = {"fails": 0, "last_fail": 0}
        return False
    return True

def breaker_record(name: str, success: bool):
    info = CIRCUIT_BREAKER.get(name, {"fails": 0, "last_fail": 0})
    if success: info["fails"] = 0
    else: info["fails"] += 1; info["last_fail"] = time.time()
    CIRCUIT_BREAKER[name] = info

if os.path.exists(HISTORY_FILE):
    try:
        with open(HISTORY_FILE) as f: DIVERGENCE_HISTORY = deque(json.load(f), maxlen=500)
    except: DIVERGENCE_HISTORY = deque(maxlen=500)
else:
    DIVERGENCE_HISTORY = deque(maxlen=500)

def save_history():
    try:
        with open(HISTORY_FILE, 'w') as f: json.dump(list(DIVERGENCE_HISTORY), f)
    except: pass

# ================= ЗАГРУЗКА ЦЕН =================
async def fetch_prices_hub(symbol: str, count: int = 100) -> Optional[np.ndarray]:
    cache_key = f"hub_{symbol}_{count}"
    now = time.time()
    if cache_key in price_cache and (now - cache_timestamps.get(cache_key, 0)) < CACHE_TTL:
        return price_cache[cache_key]

    try:
        r = await http_client.get(
            f"{HUB_URL}/candles",
            params={"symbol": symbol, "interval": "1h", "limit": count},
            timeout=10, headers=hub_headers()
        )
        if r.status_code == 200:
            candles = r.json().get("candles", [])
            if candles:
                prices = np.array([c.get("c", c.get("close", 0)) for c in candles], dtype=float)
                if len(prices) >= 20:
                    price_cache[cache_key] = prices
                    cache_timestamps[cache_key] = now
                    return prices
    except: pass
    return None

async def fetch_prices_twelvedata(symbol: str, count: int = 100) -> Optional[np.ndarray]:
    if breaker_open("twelvedata"): return None
    if not TWELVE_DATA_KEY: return None

    ticker = TICKER_MAP.get(symbol, symbol)
    try:
        url = f"https://api.twelvedata.com/time_series?symbol={ticker}&interval=1h&outputsize={count}&apikey={TWELVE_DATA_KEY}"
        r = await http_client.get(url)
        if r.status_code == 200 and 'values' in r.json():
            df = pd.DataFrame(r.json()['values']).iloc[::-1]
            prices = pd.to_numeric(df['close']).values
            breaker_record("twelvedata", True)
            return prices
    except: breaker_record("twelvedata", False)
    return None

async def fetch_prices(symbol: str, count: int = 100) -> Optional[np.ndarray]:
    prices = await fetch_prices_hub(symbol, count)
    if prices is None: prices = await fetch_prices_twelvedata(symbol, count)
    return prices

async def fetch_all_prices(symbols: List[str], count: int = 100) -> Dict[str, np.ndarray]:
    tasks = [fetch_prices(sym, count) for sym in symbols]
    results = await asyncio.gather(*tasks)
    prices = {}
    for sym, res in zip(symbols, results):
        if res is not None and len(res) >= 20: prices[sym] = res
    return prices

# ================= 🔥 РАСЧЁТ КОРРЕЛЯЦИЙ (УСИЛЕННЫЙ) =================
def compute_correlation(prices1: np.ndarray, prices2: np.ndarray, pair_name: str = "") -> Optional[Dict[str, Any]]:
    min_len = min(len(prices1), len(prices2))
    p1, p2 = prices1[-min_len:], prices2[-min_len:]
    if len(p1) < 20: return None

    returns1 = np.diff(np.log(p1))
    returns2 = np.diff(np.log(p2))
    if len(returns1) < 20: return None

    corr = np.corrcoef(returns1, returns2)[0, 1]

    # 🔥 Rolling correlation (окно 24)
    window = 24
    rolling = []
    for i in range(window, len(returns1)):
        r1 = returns1[i-window:i]; r2 = returns2[i-window:i]
        rolling.append(np.corrcoef(r1, r2)[0, 1])
    rolling = np.array(rolling)

    # 🔥 История корреляции для z-score
    if pair_name:
        if pair_name not in CORR_HISTORY: CORR_HISTORY[pair_name] = deque(maxlen=200)
        CORR_HISTORY[pair_name].append(corr)
        corr_z = calculate_zscore(corr, CORR_HISTORY[pair_name])
    else:
        corr_z = 0.0

    if len(rolling) >= 10:
        recent = rolling[-24:] if len(rolling) >= 24 else rolling
        old = rolling[:24] if len(rolling) >= 48 else rolling
        mean_old = np.mean(old)
        std_old = np.std(old) + 1e-10
        rolling_z = (corr - mean_old) / std_old

        # 🔥 Тип дивергенции
        if abs(rolling_z) > 2.5:
            if corr > 0 and rolling_z < -2.5: div_type = "STRONG_BREAKDOWN"
            elif corr < 0 and rolling_z > 2.5: div_type = "STRONG_REVERSAL"
            else: div_type = "STRONG_SHIFT"
        elif abs(rolling_z) > 2.0:
            if corr > 0 and rolling_z < -2.0: div_type = "CORRELATION_BREAKDOWN"
            elif corr < 0 and rolling_z > 2.0: div_type = "CORRELATION_REVERSAL"
            else: div_type = "CORRELATION_SHIFT"
        else:
            div_type = "NORMAL"
        
        # 🔥 Тренд корреляции
        if len(rolling) >= 48:
            half = len(rolling) // 2
            recent_trend = np.mean(rolling[-half:])
            old_trend = np.mean(rolling[:half])
            corr_trend = "INCREASING" if recent_trend > old_trend else "DECREASING" if recent_trend < old_trend else "STABLE"
        else:
            corr_trend = "STABLE"
    else:
        rolling_z = 0.0
        div_type = "NORMAL"
        corr_trend = "STABLE"

    strength = "STRONG" if abs(corr) > 0.7 else "MODERATE" if abs(corr) > 0.4 else "WEAK"
    direction = "POSITIVE" if corr > 0 else "NEGATIVE"

    return {
        "correlation": round(float(corr), 4),
        "recent_correlation": round(float(np.mean(rolling[-24:])) if len(rolling) >= 24 else corr, 4),
        "z_score": round(float(rolling_z), 2),
        "corr_zscore": round(float(corr_z), 2),
        "strength": strength,
        "direction": direction,
        "divergence": div_type,
        "trend": corr_trend,
        "is_divergence": div_type != "NORMAL"
    }

# ================= 🔥 АНАЛИЗ КОРРЕЛЯЦИЙ =================
async def analyze_correlations(symbol: str) -> Dict[str, Any]:
    pairs = CORRELATION_PAIRS.get(symbol, [])
    all_tickers = list(set([symbol] + pairs))
    prices = await fetch_all_prices(all_tickers, 100)

    if symbol not in prices:
        return {
            "correlation_score": 50.0, "signal": "WAIT", "confidence": 0.0,
            "correlations": [], "divergences": [], "divergence_count": 0,
            "signals": [], "active_pairs": 0
        }

    correlations = []
    divergences = []
    signals_list = []
    score = 50.0
    signals = {}

    for pair_symbol in pairs:
        if pair_symbol not in prices: continue
        
        pair_name = f"{symbol}/{pair_symbol}"
        corr_data = compute_correlation(prices[symbol], prices[pair_symbol], pair_name)
        
        if corr_data:
            correlations.append({"pair": pair_name, **corr_data})
            
            # 🔥 Дивергенции
            if corr_data["is_divergence"]:
                divergences.append({"pair": pair_name, **corr_data})
                severity = "HIGH" if "STRONG" in corr_data["divergence"] else "MEDIUM"
                signals_list.append({
                    "type": corr_data["divergence"],
                    "pair": pair_name,
                    "severity": severity,
                    "reason": f"Z‑score: {corr_data['z_score']}"
                })
                score -= 15 if severity == "HIGH" else 10

            # 🔥 Специальные правила для пар
            if symbol == "XAU/USD" and pair_symbol == "DXY":
                if corr_data["direction"] == "NEGATIVE" and corr_data["strength"] == "STRONG":
                    score += 6; signals["dxy"] = ("BUY", 6)
                elif corr_data["direction"] == "POSITIVE":
                    score -= 10; signals["dxy"] = ("SELL", 10)
                else:
                    signals["dxy"] = ("NEUTRAL", 0)
            
            if symbol == "XAU/USD" and pair_symbol == "TIPS":
                if corr_data["direction"] == "NEGATIVE":
                    score += 8; signals["tips"] = ("BUY", 8)
                else:
                    signals["tips"] = ("NEUTRAL", 0)
            
            if symbol == "XAU/USD" and pair_symbol == "GLD":
                if corr_data["direction"] == "POSITIVE" and corr_data["strength"] == "STRONG":
                    score += 4; signals["gld"] = ("BUY", 4)
                elif corr_data["is_divergence"]:
                    score -= 8; signals["gld"] = ("SELL", 8)
                else:
                    signals["gld"] = ("NEUTRAL", 0)
            
            if symbol == "ETH/USD" and pair_symbol == "BTC/USD":
                if corr_data["direction"] == "POSITIVE" and corr_data["strength"] == "STRONG":
                    score += 5; signals["btc"] = ("BUY", 5)
                elif corr_data["is_divergence"]:
                    score -= 10; signals["btc"] = ("SELL", 10)
                else:
                    signals["btc"] = ("NEUTRAL", 0)

    # 🔥 Множественные дивергенции = сильный сигнал
    if len(divergences) >= 3: score = max(score - 20, 3)
    elif len(divergences) >= 2: score = max(score - 12, 5)
    
    # 🔥 Нет активных пар = неопределённость
    if len(correlations) < 2: score = 50.0

    score = round(max(3, min(97, score)), 1)
    
    # 🔥 Z-score скора
    SCORE_HISTORY[symbol].append(score)
    score_z = calculate_zscore(score, SCORE_HISTORY[symbol])

    # Сигнал
    if score > 62: signal, confidence = "BUY", min(0.92, score / 100)
    elif score > 54: signal, confidence = "BUY", min(0.68, (score - 50) / 50)
    elif score < 38: signal, confidence = "SELL", min(0.92, (100 - score) / 100)
    elif score < 46: signal, confidence = "SELL", min(0.68, (50 - score) / 50)
    else: signal, confidence = "WAIT", 0.0

    if divergences:
        DIVERGENCE_HISTORY.append({
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "symbol": symbol, "divergences": divergences, "score": score
        })
        save_history()

    return {
        "correlation_score": score,
        "score_zscore": round(score_z, 2),
        "signal": signal,
        "confidence": round(confidence, 4),
        "correlations": correlations,
        "divergences": divergences,
        "divergence_count": len(divergences),
        "signals": signals_list,
        "signals_breakdown": {k: v[0] for k, v in signals.items()} if 'signals' in dir() else {},
        "active_pairs": len(correlations)
    }

# ================= ОТПРАВКА В HUB =================
async def send_signal_to_hub(symbol: str, signal: str, confidence: float, features: Dict = None):
    if features is None: features = {}
    payload = {
        "space_id": SPACE_ID, "space_name": SPACE_NAME,
        "symbol": symbol, "signal": signal, "confidence": round(confidence, 4),
        "features": features, "metadata": {"version": "4.0"},
        "timestamp": datetime.now(timezone.utc).isoformat()
    }
    for attempt in range(3):
        try:
            r = await http_client.post(f"{HUB_URL}/signals", json=payload, timeout=15, headers=hub_headers())
            if r.status_code == 200:
                logger.info(f"📤 {symbol}: {signal} conf={confidence:.3f}")
                return True
            await asyncio.sleep(2)
        except Exception as e:
            logger.warning(f"Попытка {attempt+1}: {e}")
            await asyncio.sleep(2)
    return False

# ================= ГЛАВНЫЙ СИГНАЛ =================
async def get_correlation_signal(symbol: str = "XAU/USD") -> Dict[str, Any]:
    start = time.time()
    analysis = await analyze_correlations(symbol)
    latency = int((time.time() - start) * 1000)

    features = {
        "correlation_score": analysis['correlation_score'],
        "score_zscore": analysis['score_zscore'],
        "divergence_count": analysis['divergence_count'],
        "active_pairs": analysis['active_pairs']
    }
    await send_signal_to_hub(symbol, analysis['signal'], analysis['confidence'], features)

    logger.info(f"🔗 Correlation {symbol}: {analysis['signal']} conf={analysis['confidence']:.3f} score={analysis['correlation_score']} | {latency}ms")
    return {
        "space_id": SPACE_ID, "timestamp": int(time.time()),
        "symbol": symbol, "signal": analysis['signal'],
        "confidence": analysis['confidence'], "correlation_analysis": analysis
    }

# ================= АВТО-ОТПРАВКА =================
async def auto_send_loop():
    logger.info(f"⏳ Стартовый сон {STARTUP_SLEEP}с...")
    await log_to_hub("STARTUP", f"Correlation Engine v4.0 запущен, жду {STARTUP_SLEEP}с")
    await asyncio.sleep(STARTUP_SLEEP)
    logger.info(f"🔄 Correlation Engine [интервал={AUTO_SEND_INTERVAL}с]")
    while True:
        try:
            for symbol in SYMBOLS:
                await get_correlation_signal(symbol)
                await asyncio.sleep(2)
            logger.info("✅ Correlation Engine цикл завершён")
        except Exception as e:
            logger.error(f"Ошибка: {e}")
            await log_to_hub("ERROR", f"Ошибка: {str(e)[:200]}")
        await asyncio.sleep(AUTO_SEND_INTERVAL)

# ================= FASTAPI =================
app = FastAPI(title="Tomiris Space 25 v4.0 — Correlation Engine")

@app.on_event("startup")
async def startup():
    asyncio.create_task(auto_send_loop())
    logger.info(f"🚀 Space 25 v4.0 | Хаб: {HUB_URL}")

@app.on_event("shutdown")
async def shutdown(): await http_client.aclose()

@app.get("/health")
async def health():
    return {
        "space_id": SPACE_ID, "status": "operational", "version": "4.0",
        "symbols": SYMBOLS, "history_length": len(DIVERGENCE_HISTORY)
    }

@app.head("/health")
async def health_head(): return {}

@app.get("/consilium")
async def consilium(symbol: str = Query("XAU/USD")):
    if symbol not in SYMBOLS: return {"error": "Invalid symbol"}
    return await get_correlation_signal(symbol)

@app.get("/correlation/{symbol}")
async def correlation(symbol: str):
    if symbol not in SYMBOLS: return {"error": "Invalid symbol"}
    return await analyze_correlations(symbol)

@app.get("/pair")
async def pair_correlation(s1: str = Query("XAU/USD"), s2: str = Query("DXY")):
    prices = await fetch_all_prices([s1, s2], 100)
    if s1 in prices and s2 in prices:
        corr = compute_correlation(prices[s1], prices[s2], f"{s1}/{s2}")
        return {"pair": f"{s1}/{s2}", **corr} if corr else {"error": "insufficient data"}
    return {"error": "data not available"}

@app.get("/divergences")
async def divergences(limit: int = 20): return list(DIVERGENCE_HISTORY)[-limit:]

@app.get("/all_correlations")
async def all_correlations():
    results = {}
    for sym in SYMBOLS: results[sym] = await analyze_correlations(sym)
    return results

@app.get("/send_now")
async def send_now():
    results = {}
    for symbol in SYMBOLS:
        sig = await get_correlation_signal(symbol)
        results[symbol] = sig.get("signal", "WAIT")
    return {"status": "sent", "results": results}

@app.get("/")
async def root(): return {"name": "Correlation Engine v4.0", "space_id": SPACE_ID, "hub": HUB_URL}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=7860)

print("🚀 SPACE 25 v4.0 — CORRELATION ENGINE ГОТОВ!")