Test

#!/usr/bin/env python3
# trading_setup.py
import os, math, warnings
from dataclasses import dataclass
from typing import List, Optional, Dict
import numpy as np
import pandas as pd
import yfinance as yf
import ta
from scipy.stats import rankdata

warnings.filterwarnings("ignore", category=FutureWarning)

# ---------------------------
# Config
# ---------------------------
DEFAULT_LOOKBACK_YEARS = 2
MIN_HISTORY_DAYS = 252
MOM_WINDOW_DAYS = {"1m":21, "3m":63, "6m":126, "12m":252}
TECH_EMAS = (20, 50, 200)
WEIGHTS = {
    "technical_flow": 0.35,
    "options_sentiment": 0.20,
    "dark_pool": 0.15,
    "momentum": 0.20,
    "growth": 0.10,
}

# ---------------------------
# Utils
# ---------------------------
def winsorize_series(s: pd.Series, p: float = 0.01) -> pd.Series:
    if s.dropna().empty: return s
    lo, hi = s.quantile(p), s.quantile(1-p)
    return s.clip(lower=lo, upper=hi)

def scale_0_100(s: pd.Series) -> pd.Series:
    if s.dropna().nunique() <= 1: 
        return pd.Series(50.0, index=s.index)  # flat middle if no variance
    s_w = winsorize_series(s)
    lo, hi = s_w.min(), s_w.max()
    return (100*(s_w - lo)/(hi - lo)).astype(float)

def pct_change(df: pd.DataFrame, col="Adj Close", window=21) -> float:
    if len(df) < window+1: return np.nan
    return df[col].iloc[-1] / df[col].iloc[-1-window] - 1.0

def safe_div(a, b):
    try:
        return a / b if b and not np.isnan(b) else np.nan
    except ZeroDivisionError:
        return np.nan

# ---------------------------
# Data classes
# ---------------------------
@dataclass
class DarkPoolRecord:
    ticker: str
    dp_net_flow: Optional[float] = None
    dp_score: Optional[float] = None  # if already 0-100

@dataclass
class FundamentalsRecord:
    ticker: str
    revenue_yoy: Optional[float] = None
    eps_yoy: Optional[float] = None

# ---------------------------
# Fetchers
# ---------------------------
def fetch_price_history(ticker: str, years: int = DEFAULT_LOOKBACK_YEARS) -> Optional[pd.DataFrame]:
    try:
        df = yf.Ticker(ticker).history(period=f"{years}y", auto_adjust=False)
        if df is None or df.empty or len(df) < MIN_HISTORY_DAYS//2:
            return None
        return df
    except Exception:
        return None

def fetch_option_chain_put_call_ratio(ticker: str) -> Optional[float]:
    """
    Basic net options sentiment via Put/Call volume ratio (lower = bullish).
    Returns PCR; we'll map to a 0-100 bullish score later.
    """
    try:
        tk = yf.Ticker(ticker)
        exps = tk.options
        if not exps: return None
        # choose nearest non-expired expiration
        exp = exps[0]
        chain = tk.option_chain(exp)
        calls, puts = chain.calls, chain.puts
        if calls is None or puts is None or calls.empty or puts.empty:
            return None
        call_vol = calls["volume"].replace(0, np.nan).sum(skipna=True)
        put_vol  = puts["volume"].replace(0, np.nan).sum(skipna=True)
        pcr = safe_div(put_vol, call_vol)
        return float(pcr) if pcr is not None else None
    except Exception:
        return None

def load_dark_pool_csv(path: str) -> Dict[str, DarkPoolRecord]:
    if not path or not os.path.exists(path): return {}
    df = pd.read_csv(path)
    out = {}
    for _, r in df.iterrows():
        out[str(r["ticker"]).upper()] = DarkPoolRecord(
            ticker=str(r["ticker"]).upper(),
            dp_net_flow=(r["dp_net_flow"] if "dp_net_flow" in df.columns else None),
            dp_score=(r["dp_score"] if "dp_score" in df.columns else None),
        )
    return out

def load_fundamentals_csv(path: str) -> Dict[str, FundamentalsRecord]:
    if not path or not os.path.exists(path): return {}
    df = pd.read_csv(path)
    out = {}
    for _, r in df.iterrows():
        out[str(r["ticker"]).upper()] = FundamentalsRecord(
            ticker=str(r["ticker"]).upper(),
            revenue_yoy=(r["revenue_yoy"] if "revenue_yoy" in df.columns else None),
            eps_yoy=(r["eps_yoy"] if "eps_yoy" in df.columns else None),
        )
    return out

# ---------------------------
# Scoring pillars
# ---------------------------
def technical_flow_score(df: pd.DataFrame) -> float:
    """
    0-100 score from EMAs stack, RSI, MACD, ADX, Volume trend.
    """
    try:
        close = df["Adj Close"].rename("close") if "Adj Close" in df else df["Close"].rename("close")
        high, low, vol = df["High"], df["Low"], df["Volume"]

        ema20 = close.ewm(span=TECH_EMAS[0]).mean()
        ema50 = close.ewm(span=TECH_EMAS[1]).mean()
        ema200= close.ewm(span=TECH_EMAS[2]).mean()

        rsi = ta.momentum.RSIIndicator(close, window=14).rsi()
        macd = ta.trend.MACD(close).macd_diff()
        adx = ta.trend.ADXIndicator(high, low, close).adx()

        # Normalize components
        # Trend stack (0-40)
        stack_score = (
            (1 if ema20.iloc[-1] > ema50.iloc[-1] else 0) +
            (1 if ema50.iloc[-1] > ema200.iloc[-1] else 0) +
            (1 if ema20.iloc[-1] > ema200.iloc[-1] else 0)
        ) / 3.0 * 40.0

        # RSI favor 50-70 (bullish), scale to 0-30
        rsi_now = rsi.iloc[-1]
        rsi_component = max(0.0, min(30.0, (rsi_now - 30) / 40 * 30)) if not np.isnan(rsi_now) else 15.0

        # MACD histogram latest vs distribution (0-15)
        macd_scaled = scale_0_100(macd).iloc[-1] if macd.notna().any() else 50.0
        macd_component = macd_scaled * 0.15

        # ADX trend strength (25+ is trending) → 0-15
        adx_now = adx.iloc[-1]
        adx_component = max(0.0, min(15.0, (adx_now - 15) / 20 * 15)) if not np.isnan(adx_now) else 7.5

        # Volume trend (20d vs 200d)
        vol20 = vol.rolling(20).mean().iloc[-1]
        vol200= vol.rolling(200).mean().iloc[-1]
        vol_ratio = safe_div(vol20, vol200)
        vol_component = 0 if vol_ratio is None or np.isnan(vol_ratio) else max(0.0, min(10.0, (vol_ratio-0.8)/0.7*10))

        score = stack_score + rsi_component + macd_component + adx_component + vol_component
        return float(max(0.0, min(100.0, score)))
    except Exception:
        return np.nan

def options_sentiment_score(pcr: Optional[float]) -> float:
    """
    Map Put/Call ratio to bullish 0-100.
    Typical PCR ranges: ~0.5-1.5. Lower => more bullish.
    """
    if pcr is None or np.isnan(pcr): return np.nan
    # Clip to [0.3, 2.0] then invert
    pcr_c = max(0.3, min(2.0, pcr))
    # 0.3 -> ~100, 1.0 -> 50, 2.0 -> ~0
    score = (2.0 - pcr_c) / (2.0 - 0.3) * 100.0
    return float(max(0.0, min(100.0, score)))

def momentum_scores(df: pd.DataFrame) -> Dict[str, float]:
    out = {}
    for label, win in MOM_WINDOW_DAYS.items():
        out[f"mom_{label}"] = pct_change(df, window=win)
    # Composite momentum (skip 1m if you prefer medium-term)
    comp = 0.1*out.get("mom_1m", np.nan) + 0.3*out.get("mom_3m", np.nan) + 0.25*out.get("mom_6m", np.nan) + 0.35*out.get("mom_12m", np.nan)
    out["mom_composite_raw"] = comp
    return out

def growth_score(revenue_yoy: Optional[float], eps_yoy: Optional[float], peers_df: pd.DataFrame) -> float:
    """
    Rank/percentile within peers for revenue & EPS YoY → average → 0-100.
    If missing, returns NaN.
    """
    if revenue_yoy is None or eps_yoy is None or np.isnan(revenue_yoy) or np.isnan(eps_yoy):
        return np.nan
    # Build temp arrays with peers + this point for robust scaling
    rev = peers_df["revenue_yoy"].astype(float)
    eps = peers_df["eps_yoy"].astype(float)
    rev2 = pd.concat([rev, pd.Series([revenue_yoy])], ignore_index=True)
    eps2 = pd.concat([eps, pd.Series([eps_yoy])], ignore_index=True)
    rev_p = rankdata(rev2, method="average")/len(rev2)*100
    eps_p = rankdata(eps2, method="average")/len(eps2)*100
    # last element corresponds to our ticker
    return float(np.nanmean([rev_p[-1], eps_p[-1]]))

def dark_pool_score(rec: Optional[DarkPoolRecord], peer_records: List[DarkPoolRecord]) -> float:
    if rec is None: return np.nan
    if rec.dp_score is not None and not np.isnan(rec.dp_score):
        return float(max(0.0, min(100.0, rec.dp_score)))
    # derive from dp_net_flow vs peers
    if rec.dp_net_flow is None or np.isnan(rec.dp_net_flow): return np.nan
    vals = pd.Series([r.dp_net_flow for r in peer_records if r.dp_net_flow is not None and not np.isnan(r.dp_net_flow)])
    if vals.empty: 
        return np.nan
    vals = winsorize_series(vals)
    lo, hi = vals.min(), vals.max()
    sc = (rec.dp_net_flow - lo) / (hi - lo) * 100 if hi > lo else 50.0
    return float(max(0.0, min(100.0, sc)))

# ---------------------------
# Orchestrator
# ---------------------------
def analyze_tickers(
    tickers: List[str],
    dark_pool_csv: Optional[str] = None,
    fundamentals_csv: Optional[str] = None
) -> pd.DataFrame:

    tickers = [t.upper() for t in tickers]
    dp_map = load_dark_pool_csv(dark_pool_csv) if dark_pool_csv else {}
    f_map  = load_fundamentals_csv(fundamentals_csv) if fundamentals_csv else {}

    # Pre-collect peer lists for scaling DP/Growth
    dp_peers = list(dp_map.values())
    growth_peers = pd.DataFrame(
        [{"ticker":k, "revenue_yoy":v.revenue_yoy, "eps_yoy":v.eps_yoy}
         for k,v in f_map.items() if v.revenue_yoy is not None and v.eps_yoy is not None]
    )
    if growth_peers.empty:
        growth_peers = pd.DataFrame({"ticker":[], "revenue_yoy":[], "eps_yoy":[]})

    rows = []
    # Pass 1: compute price-based things
    cache_price = {}
    for t in tickers:
        df = fetch_price_history(t)
        cache_price[t] = df

    # Pass 2: compute pillar scores
    # Momentum raw scores for all (for cross-sectional scaling)
    raw_moms = {}
    for t in tickers:
        df = cache_price.get(t)
        raw_moms[t] = np.nan
        if df is not None:
            ms = momentum_scores(df)
            raw_moms[t] = ms["mom_composite_raw"]

    # Scale momentum cross-sectionally
    mom_scaled = scale_0_100(pd.Series(raw_moms))
    # Gather outputs
    for t in tickers:
        df = cache_price.get(t)
        tech = technical_flow_score(df) if df is not None else np.nan
        pcr = fetch_option_chain_put_call_ratio(t)
        opt = options_sentiment_score(pcr)

        dp_rec = dp_map.get(t)
        dp_sc = dark_pool_score(dp_rec, dp_peers)

        # growth
        f_rec = f_map.get(t, FundamentalsRecord(ticker=t, revenue_yoy=np.nan, eps_yoy=np.nan))
        gr_sc = growth_score(f_rec.revenue_yoy, f_rec.eps_yoy, growth_peers) if not growth_peers.empty else np.nan

        # final composite (handle NaNs by redistributing weights over available pillars)
        components = {
            "technical_flow": tech,
            "options_sentiment": opt,
            "dark_pool": dp_sc,
            "momentum": mom_scaled.get(t, np.nan),
            "growth": gr_sc
        }
        avail = {k:v for k,v in components.items() if v is not None and not np.isnan(v)}
        if not avail:
            composite = np.nan
        else:
            w = {k:WEIGHTS[k] for k in avail.keys()}
            w_sum = sum(w.values())
            w_norm = {k:v/w_sum for k,v in w.items()}
            composite = sum(avail[k]*w_norm[k] for k in avail.keys())

        # add a simple signal
        signal = (
            "Strong Buy" if composite >= 80 else
            "Buy"        if composite >= 65 else
            "Watch"      if composite >= 50 else
            "Avoid"
        )

        # momentum components for visibility
        mom_parts = momentum_scores(df) if df is not None else {}
        rows.append({
            "ticker": t,
            "technical_flow": round(tech,2) if not np.isnan(tech) else np.nan,
            "options_pcr": round(pcr,3) if pcr is not None and not np.isnan(pcr) else np.nan,
            "options_sentiment": round(opt,2) if not np.isnan(opt) else np.nan,
            "dark_pool": round(dp_sc,2) if not np.isnan(dp_sc) else np.nan,
            "mom_1m": mom_parts.get("mom_1m", np.nan),
            "mom_3m": mom_parts.get("mom_3m", np.nan),
            "mom_6m": mom_parts.get("mom_6m", np.nan),
            "mom_12m": mom_parts.get("mom_12m", np.nan),
            "momentum": round(mom_scaled.get(t, np.nan),2) if not np.isnan(mom_scaled.get(t, np.nan)) else np.nan,
            "growth": round(gr_sc,2) if not np.isnan(gr_sc) else np.nan,
            "composite": round(composite,2) if not np.isnan(composite) else np.nan,
            "signal": signal
        })

    out = pd.DataFrame(rows)
    # tidy momentum pct columns
    for c in ["mom_1m","mom_3m","mom_6m","mom_12m"]:
        if c in out.columns:
            out[c] = (out[c]*100).round(2)
    out = out.sort_values("composite", ascending=False, na_position="last").reset_index(drop=True)
    return out

# ---------------------------
# CLI
# ---------------------------
if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="Net Options Sentiment + Dark Pool + Technical Flow + Momentum + Growth ranker")
    parser.add_argument("--tickers", nargs="+", required=True, help="e.g., AAPL MSFT NVDA")
    parser.add_argument("--dark_pool_csv", type=str, default=None, help="Optional path to dark pool CSV (ticker,dp_net_flow,dp_score)")
    parser.add_argument("--fundamentals_csv", type=str, default=None, help="Optional path to fundamentals CSV (ticker,revenue_yoy,eps_yoy)")
    parser.add_argument("--out_csv", type=str, default="trading_setup_output.csv", help="Output CSV path")
    args = parser.parse_args()

    df = analyze_tickers(args.tickers, args.dark_pool_csv, args.fundamentals_csv)
    pd.set_option("display.max_columns", None)
    print(df.to_string(index=False))
    df.to_csv(args.out_csv, index=False)
    print(f"\nSaved: {args.out_csv}")