app.py

# ===============================================================
#  Forex EMA + Dynamic Normalization API (Model B)
#  Versi Final – Aman untuk Hugging Face Spaces (tanpa cache)
# ===============================================================

from fastapi import FastAPI
from pydantic import BaseModel
import pandas as pd
import numpy as np
import yfinance as yf
from datetime import datetime, timedelta
import logging
import tempfile
import os

# ===============================================================
# Konfigurasi Logging
# ===============================================================
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s — %(levelname)s — %(message)s"
)
logger = logging.getLogger(__name__)

# ===============================================================
# Konfigurasi FastAPI
# ===============================================================
app = FastAPI(
    title="Model B – EMA & Dynamic Scaling API",
    description="API untuk menghitung EMA, normalisasi, dan analisis tren otomatis berdasarkan data yfinance",
    version="2.3"
)

PAIR = "EURUSD=X"
BASE_WINDOW = 60

# Matikan cache yfinance agar tidak menulis ke /.cache
os.environ["YFINANCE_CACHE_DISABLE"] = "1"
os.environ["YFINANCE_NO_TZ_CACHE"] = "1"
yf.set_tz_cache_location(tempfile.gettempdir())

# ===============================================================
# Schema
# ===============================================================
class DateRange(BaseModel):
    start_date: str
    end_date: str


# ===============================================================
# Helper Function – Load data langsung dari yfinance
# ===============================================================
def load_yf_data(pair, start, end):
    """
    Ambil data yfinance tanpa cache, hanya return kolom ['date', 'close'].
    """
    try:
        df = yf.download(pair, start=start, end=end, auto_adjust=True, progress=False)

        if df.empty:
            raise ValueError("YFinance gagal mengambil data, data kosong.")

        # Jika MultiIndex, ambil level pertama
        if isinstance(df.columns, pd.MultiIndex):
            df.columns = df.columns.get_level_values(0)

        # Ambil hanya kolom Close
        close_col = [c for c in df.columns if "close" in c.lower()]
        if not close_col:
            raise ValueError(f"Tidak ada kolom 'Close' ditemukan di {df.columns.tolist()}")

        df = df.reset_index()[["Date", close_col[0]]]
        df.columns = ["date", "close"]
        df["date"] = pd.to_datetime(df["date"]).dt.date
        df["close"] = pd.to_numeric(df["close"], errors="coerce")
        df = df.dropna(subset=["close"]).reset_index(drop=True)

        logger.info(f"✅ Berhasil ambil data {len(df)} baris dari {pair}")
        return df

    except Exception as e:
        logger.error(f"❌ Error load_yf_data(): {e}", exc_info=True)
        raise ValueError(str(e))

    logger.info(f"Requested summary start={start_date.date()} end={end_date.date()}")
    logger.info(f"Loaded rows from yfinance: {len(df)}")
    
    if len(df) < 50:
        return {
          "status": "error",
          "message": "Data terlalu sedikit (butuh minimal 50 hari).",
          "data_points": len(df)
        }

# ===============================================================
# Helper Function – Manual EMA
# ===============================================================
def ema_manual(prices, span):
    if len(prices) < span:
        return [np.nan] * len(prices)

    ema = [np.nan] * len(prices)
    alpha = 2 / (span + 1)
    ema[span - 1] = np.mean(prices[:span])

    for i in range(span, len(prices)):
        ema[i] = alpha * prices[i] + (1 - alpha) * ema[i - 1]

    return ema


# ===============================================================
# Helper Function – Dynamic Scaling
# ===============================================================
def get_dynamic_minmax():
    today = datetime.now().date()
    start = today - timedelta(days=BASE_WINDOW)
    df = load_yf_data(PAIR, start, today + timedelta(days=1))
    close_min = df["close"].min()
    close_max = df["close"].max()
    logger.info(f"Dynamic Min/Max Close: {close_min:.5f} / {close_max:.5f}")
    return float(close_min), float(close_max)


def normalize_close(value, close_min, close_max):
    if close_max == close_min:
        return 0.5
    return (value - close_min) / (close_max - close_min)


# ===============================================================
# Helper Function – Trend Analysis
# ===============================================================
def analyze_trend(latest_row):
    ema20 = latest_row["EMA20"]
    ema50 = latest_row["EMA50"]
    close = latest_row["close"]

    if ema20 > ema50:
        trend = "bullish"
    elif ema20 < ema50:
        trend = "bearish"
    else:
        trend = "neutral"

    diff = abs(ema20 - ema50) / ema50 * 100 if ema50 != 0 else 0
    if diff > 0.3:
        strength = "strong"
    elif diff > 0.1:
        strength = "moderate"
    else:
        strength = "weak"

    if close > ema20 and close > ema50:
        price_position = "above both EMA — possible continuation"
    elif close < ema20 and close < ema50:
        price_position = "below both EMA — possible correction"
    else:
        price_position = "between EMAs — indecision zone"

    return {
        "trend": trend,
        "strength": strength,
        "price_position": price_position,
        "ema_gap_percent": round(diff, 3)
    }


# ===============================================================
# Endpoint: /analyze
# ===============================================================
@app.post("/analyze")
def analyze_ema_endpoint(input_data: DateRange):
    try:
        start_date = pd.to_datetime(input_data.start_date)
        end_date = pd.to_datetime(input_data.end_date)
        extended_start = start_date - timedelta(days=60)

        df = load_yf_data(PAIR, extended_start, end_date + timedelta(days=1))
        if len(df) < 50:
            return {"status": "error", "message": "Data terlalu sedikit (butuh minimal 50 hari)."}

        df["EMA20"] = ema_manual(df["close"].values.tolist(), 20)
        df["EMA50"] = ema_manual(df["close"].values.tolist(), 50)
        df = df.dropna().reset_index(drop=True)

        close_min, close_max = get_dynamic_minmax()
        df["norm_close"] = df["close"].apply(lambda x: normalize_close(x, close_min, close_max))

        chart_data = {
            "dates": [str(d) for d in df["date"].tolist()],
            "close": df["close"].round(6).tolist(),
            "EMA20": df["EMA20"].round(6).tolist(),
            "EMA50": df["EMA50"].round(6).tolist(),
            "norm_close": df["norm_close"].round(6).tolist(),
            "min_close": close_min,
            "max_close": close_max
        }

        return {
            "status": "ok",
            "pair": PAIR,
            "start_date": str(start_date.date()),
            "end_date": str(end_date.date()),
            "data_points": len(df),
            "chart_data": chart_data
        }

    except Exception as e:
        logger.error(f"Error di /analyze: {e}", exc_info=True)
        return {"status": "error", "message": str(e)}


# ===============================================================
# Endpoint: /summary
# ===============================================================
@app.post("/summary")
def ema_summary_endpoint(input_data: DateRange):
    try:
        start_date = pd.to_datetime(input_data.start_date)
        end_date = pd.to_datetime(input_data.end_date)
        extended_start = start_date - timedelta(days=60)

        df = load_yf_data(PAIR, extended_start, end_date + timedelta(days=1))
        if len(df) < 50:
            return {"status": "error", "message": "Data terlalu sedikit (butuh minimal 50 hari)."}

        df["EMA20"] = ema_manual(df["close"].values.tolist(), 20)
        df["EMA50"] = ema_manual(df["close"].values.tolist(), 50)
        df = df.dropna().reset_index(drop=True)

        latest = df.iloc[-1]
        analysis = analyze_trend(latest)

        return {
            "status": "ok",
            "pair": PAIR,
            "as_of_date": str(latest["date"]),
            "close": round(float(latest["close"]), 6),
            "EMA20": round(float(latest["EMA20"]), 6),
            "EMA50": round(float(latest["EMA50"]), 6),
            "trend_analysis": analysis
        }

    except Exception as e:
        logger.error(f"Error di /summary: {e}", exc_info=True)
        return {"status": "error", "message": str(e)}


# ===============================================================
# Root Test
# ===============================================================
@app.get("/")
def root():
    return {"message": "Model B API (EMA + Trend Summary) aktif 🚀"}