app.py

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

REQUIRED_PACKAGES = {
    'numpy': 'numpy',
    'pandas': 'pandas',
    'httpx': 'httpx',
    '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 28 v2.2 — MARKET REGIME & BUBBLE SENTINEL (АВТО-ОТПРАВКА)
# ============================================
import os, time, json, logging, asyncio
from typing import Dict, Any, List, Optional
from datetime import datetime, timezone
from collections import deque
import numpy as np
import pandas as pd
import httpx
from fastapi import FastAPI, Query

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

# ================= КОНФИГУРАЦИЯ =================
SYMBOLS = ["XAU/USD", "ETH/USD", "SOL/USD"]
HUB_URL = os.getenv("SPACE17_URL", "https://tomiris-ai-name5-5.hf.space")
SPACE9_URL = os.getenv("SPACE9_URL", "https://nuxotetotnicksvoboden-name3.hf.space")
SPACE22_URL = os.getenv("SPACE22_URL", "https://tomirisai80-tomirisanal4.hf.space")
SPACE26_URL = os.getenv("SPACE26_URL", "https://tomirisg25-tomirisgold2.hf.space")
TWELVE_DATA_KEY = os.getenv("TWELVE_DATA_KEY", "e3740c072fda4fe8b8539d40b07e445e")

# Интервал авто-отправки
AUTO_SEND_INTERVAL = int(os.getenv("AUTO_SEND_INTERVAL", "300"))

BUBBLE_HISTORY_FILE = "bubble_history.json"
CACHE_TTL = {"candles": 300, "onchain": 600, "sentiment": 300, "derivatives": 120}

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

# ================= ГЛОБАЛЬНЫЙ КЭШ И BREAKER =================
cache_store = {}
cache_times = {}
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(BUBBLE_HISTORY_FILE):
    try:
        with open(BUBBLE_HISTORY_FILE) as f: BUBBLE_HISTORY = deque(json.load(f), maxlen=500)
    except: BUBBLE_HISTORY = deque(maxlen=500)
else: BUBBLE_HISTORY = deque(maxlen=500)

def save_bubble_history():
    with open(BUBBLE_HISTORY_FILE, 'w') as f: json.dump(list(BUBBLE_HISTORY), f)

# ================= ЗАГРУЗКА ДАННЫХ =================
async def fetch_candles(symbol: str, tf: str = "1h", count: int = 200) -> Optional[pd.DataFrame]:
    cache_key = f"candles_{symbol}_{tf}_{count}"
    if cache_key in cache_store and time.time() - cache_times.get(cache_key, 0) < CACHE_TTL["candles"]:
        return cache_store[cache_key]
    if breaker_open("hub"): return None
    try:
        r = await http_client.get(f"{HUB_URL}/candles", params={"symbol": symbol, "interval": tf, "limit": count})
        if r.status_code == 200:
            data = r.json().get("candles", [])
            if data:
                df = pd.DataFrame(data)
                df.rename(columns={"o": "open", "h": "high", "l": "low", "c": "close", "v": "volume", "t": "timestamp"}, inplace=True)
                df["close"] = pd.to_numeric(df["close"], errors="coerce")
                df["high"] = pd.to_numeric(df["high"], errors="coerce")
                df["low"] = pd.to_numeric(df["low"], errors="coerce")
                if "volume" in df.columns: df["volume"] = pd.to_numeric(df["volume"], errors="coerce").fillna(0)
                breaker_record("hub", True)
                cache_store[cache_key] = df; cache_times[cache_key] = time.time()
                return df
        breaker_record("hub", False)
    except: breaker_record("hub", False)
    return None

async def fetch_onchain_volume(symbol: str) -> Optional[float]:
    if not SPACE9_URL or breaker_open("space9"): return None
    try:
        r = await http_client.get(f"{SPACE9_URL}/consilium?symbol={symbol}")
        if r.status_code == 200:
            data = r.json()
            metrics = data.get("onchain_analysis", {}).get("metrics", {})
            network = metrics.get("network", {})
            vol = network.get("tx_volume_24h") or network.get("total_volume_24h")
            if vol: breaker_record("space9", True); return float(vol)
    except: breaker_record("space9", False)
    return None

async def fetch_sentiment_signal(symbol: str) -> Optional[Dict]:
    if not SPACE22_URL or breaker_open("space22"): return None
    try:
        r = await http_client.get(f"{SPACE22_URL}/sentiment/{symbol}")
        if r.status_code == 200: breaker_record("space22", True); return r.json()
    except: breaker_record("space22", False)
    return None

async def fetch_derivatives(symbol: str) -> Optional[Dict]:
    if not SPACE26_URL or breaker_open("space26"): return None
    try:
        r = await http_client.get(f"{SPACE26_URL}/consilium?symbol={symbol}")
        if r.status_code == 200:
            data = r.json()
            deriv = data.get("derivative_analysis", {}).get("metrics", {})
            breaker_record("space26", True); return deriv
    except: breaker_record("space26", False)
    return None

# ================= ИНДИКАТОРЫ =================
def safe_rsi(close: pd.Series, period: int = 14) -> float:
    try:
        delta = close.diff()
        gain = delta.clip(lower=0).rolling(period, min_periods=period).mean()
        loss = (-delta.clip(upper=0)).rolling(period, min_periods=period).mean()
        rs = gain / (loss + 1e-10)
        return float(100 - (100 / (1 + rs.iloc[-1])))
    except: return 50.0

def calculate_euphoria_index(df: pd.DataFrame, funding_signal: Optional[str] = None,
                             oi_change: Optional[float] = None, sentiment_fear: Optional[float] = None) -> float:
    if df is None or len(df) < 20: return 50.0
    close = df['close']; volume = df['volume'] if 'volume' in df.columns else pd.Series([1.0]*len(df))
    score = 0.0
    rsi = safe_rsi(close, 14)
    if rsi > 80: score += 25
    elif rsi > 70: score += 18
    elif rsi > 60: score += 8
    elif rsi < 30: score -= 15
    elif rsi < 20: score -= 20
    if len(close) >= 50:
        sma50 = close.rolling(50).mean().iloc[-1]
        dev = (close.iloc[-1] - sma50) / sma50 * 100
        if dev > 20: score += 20
        elif dev > 10: score += 12
        elif dev > 5: score += 5
        elif dev < -20: score -= 15
    if len(volume) >= 20:
        avg_vol = volume.rolling(20).mean().iloc[-1]
        vol_ratio = volume.iloc[-1] / (avg_vol + 1e-10)
        if vol_ratio > 3: score += 15
        elif vol_ratio > 2: score += 8
    if funding_signal:
        if funding_signal in ('EXTREME_LONG', 'BEARISH'): score += 15
        elif funding_signal == 'BULLISH': score -= 10
    if oi_change:
        if oi_change > 10: score += 10
        elif oi_change < -10: score -= 5
    if sentiment_fear is not None:
        greed = 100 - sentiment_fear
        if greed > 70: score += 10
        elif greed < 30: score -= 10
    return max(0.0, min(100.0, score))

def calculate_nvt_ratio(symbol: str, df: pd.DataFrame, onchain_volume: Optional[float] = None) -> float:
    if df is None or len(df) < 24: return 0.0
    close = df['close'].iloc[-1]
    supply = {"ETH/USD": 120_000_000, "SOL/USD": 440_000_000, "XAU/USD": 1}.get(symbol, 1)
    market_cap = close * supply
    if onchain_volume and onchain_volume > 0: nvt = market_cap / onchain_volume
    else:
        volume = df['volume'] if 'volume' in df.columns else pd.Series([1.0]*len(df))
        daily_volume = volume.iloc[-24:].sum() if len(volume) >= 24 else volume.sum()
        if daily_volume <= 0: return 0.0
        nvt = market_cap / daily_volume
    if "ETH" in symbol: normal_low, normal_high = 30, 100
    elif "SOL" in symbol: normal_low, normal_high = 50, 150
    else: return 0.0
    if nvt > normal_high: return min(30, (nvt - normal_high) / 10)
    elif nvt < normal_low: return max(-20, (nvt - normal_low) / 10)
    return 0.0

# ================= РЕЖИМ РЫНКА =================
async def detect_market_regime(symbol: str) -> Dict[str, Any]:
    df = await fetch_candles(symbol, "1h", 200)
    if df is None or len(df) < 50: return {"regime": "UNKNOWN", "bubble_probability": 0, "veto": False, "euphoria_index": 50}
    onchain_vol = None if symbol == "XAU/USD" else await fetch_onchain_volume(symbol)
    deriv = None if symbol == "XAU/USD" else await fetch_derivatives(symbol)
    sent = await fetch_sentiment_signal(symbol)
    funding_signal = None; oi_change = None
    if deriv:
        funding = deriv.get("funding_rate", {}); funding_signal = funding.get("signal")
        oi = deriv.get("open_interest", {}); oi_change = oi.get("change_pct")
    fear_ratio = None
    if sent:
        reddit = sent.get("metrics", {}).get("reddit", {})
        fear_ratio = reddit.get("fear_ratio") or sent.get("fear_ratio")
    close = df['close'].astype(float)
    returns = np.diff(np.log(close.values))
    volatility = float(np.std(returns[-24:])) if len(returns) >= 24 else 0.01
    euphoria = calculate_euphoria_index(df, funding_signal, oi_change, fear_ratio)
    nvt_score = 0.0 if symbol == "XAU/USD" else calculate_nvt_ratio(symbol, df, onchain_vol)
    if euphoria > 70 or nvt_score > 20:
        regime = "BUBBLE"; bubble_probability = min(100, euphoria + nvt_score * 2); veto = True; signal = "FORCE_WAIT"
    elif euphoria > 55: regime = "EUPHORIA"; bubble_probability = euphoria; veto = False; signal = "CAUTION"
    elif volatility > 0.03: regime = "VOLATILE"; bubble_probability = 40; veto = False; signal = "NEUTRAL"
    elif close.iloc[-1] < close.iloc[-50] * 0.8 and euphoria < 30: regime = "CAPITULATION"; bubble_probability = 10; veto = False; signal = "NORMAL"
    elif abs(close.iloc[-1] - close.iloc[-20]) / close.iloc[-20] < 0.02: regime = "RANGE"; bubble_probability = 20; veto = False; signal = "NEUTRAL"
    else: regime = "TREND"; bubble_probability = max(0, min(100, 50 + (close.iloc[-1] > close.iloc[-50] and 15 or -15))); veto = False; signal = "NORMAL"
    BUBBLE_HISTORY.append({"timestamp": datetime.now(timezone.utc).isoformat(), "symbol": symbol, "regime": regime, "bubble_probability": bubble_probability, "euphoria": euphoria})
    save_bubble_history()
    return {"regime": regime, "bubble_probability": round(bubble_probability, 1), "euphoria_index": round(euphoria, 1), "nvt_score": round(nvt_score, 1), "rsi_14": round(safe_rsi(close, 14), 1), "volatility_24h_pct": round(volatility * 100, 3), "veto": veto, "signal": signal}

# ================= ОТПРАВКА В HUB =================
async def send_signal_to_hub(symbol: str, direction: str, confidence: float):
    try:
        resp = await http_client.post(f"{HUB_URL}/signal", json={
            "space": "space_28_regime", "symbol": symbol,
            "direction": direction, "confidence": confidence,
            "raw": json.dumps({"source": "space_28_regime"})
        }, timeout=10)
        if resp.status_code == 200: logger.info(f"📤 {symbol}: {direction} conf={confidence:.3f} отправлен в Hub")
        else: logger.warning(f"Hub вернул {resp.status_code}")
    except Exception as e: logger.error(f"Ошибка отправки в Hub: {e}")

# ================= ГЛАВНЫЙ СИГНАЛ =================
async def get_regime_signal(symbol: str = "XAU/USD") -> Dict[str, Any]:
    start = time.time(); regime_data = await detect_market_regime(symbol)
    latency = int((time.time() - start) * 1000)
    direction = "WAIT" if regime_data["veto"] else "NEUTRAL"
    confidence = regime_data["bubble_probability"] / 100 if regime_data["veto"] else 0.0
    await send_signal_to_hub(symbol, direction, confidence)
    result = {"space": "space_28_regime", "timestamp": int(time.time()), "symbol": symbol, "signal": {"direction": direction, "confidence": confidence, "veto": regime_data["veto"], "veto_reason": "BUBBLE_DETECTED" if regime_data["veto"] else None}, "regime_analysis": regime_data, "latency_ms": latency}
    logger.info(f"🫧 Regime {symbol}: {regime_data['regime']} veto={regime_data['veto']}")
    return result

# ================= АВТО-ОТПРАВКА =================
async def auto_send_loop():
    logger.info(f"🔄 Авто-отправка Regime Sentinel запущена (интервал {AUTO_SEND_INTERVAL}с)")
    await asyncio.sleep(30)
    while True:
        try:
            for symbol in SYMBOLS: await get_regime_signal(symbol); await asyncio.sleep(2)
            logger.info("✅ Regime Sentinel авто-отправка завершена")
        except Exception as e: logger.error(f"Ошибка авто-отправки: {e}")
        await asyncio.sleep(AUTO_SEND_INTERVAL)

# ================= FASTAPI =================
app = FastAPI(title="Tomiris Space 28 v2.2 — Market Regime & Bubble Sentinel (Auto-Hub)")

@app.on_event("startup")
async def startup():
    asyncio.create_task(auto_send_loop())
    logger.info("🚀 Space 28 v2.2 запущен с авто-отправкой в Hub")

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

@app.get("/health")
async def health():
    return {"status": "operational", "version": "2.2", "hub_url": HUB_URL, "auto_send_interval": AUTO_SEND_INTERVAL, "features": ["Real NVT", "Composite Bubble Index", "Funding/Sentiment Integration"]}

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

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

@app.get("/euphoria/{symbol}")
async def euphoria(symbol: str):
    if symbol not in SYMBOLS: return {"error": "Invalid symbol"}
    df = await fetch_candles(symbol)
    if df is None: return {"error": "no data"}
    deriv = await fetch_derivatives(symbol) if symbol != "XAU/USD" else None
    sent = await fetch_sentiment_signal(symbol)
    fear = None
    if sent:
        reddit = sent.get("metrics", {}).get("reddit", {})
        fear = reddit.get("fear_ratio")
    funding_signal = deriv.get("funding_rate", {}).get("signal") if deriv else None
    oi_change = deriv.get("open_interest", {}).get("change_pct") if deriv else None
    ei = calculate_euphoria_index(df, funding_signal, oi_change, fear)
    return {"symbol": symbol, "euphoria_index": round(ei, 1)}

@app.get("/nvt/{symbol}")
async def nvt(symbol: str):
    if symbol not in ["ETH/USD", "SOL/USD"]: return {"error": "NVT доступен только для крипты"}
    df = await fetch_candles(symbol)
    if df is None: return {"error": "no data"}
    onchain_vol = await fetch_onchain_volume(symbol)
    nvt_score = calculate_nvt_ratio(symbol, df, onchain_vol)
    return {"symbol": symbol, "nvt_score": round(nvt_score, 1)}

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

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

print("🚀 SPACE 28 v2.2 — MARKET REGIME & BUBBLE SENTINEL (АВТО-ОТПРАВКА) ЗАПУЩЕН!")