import gradio as gr
import requests
import pandas as pd
import numpy as np
from datetime import datetime
import time
# ──────────────────────────────────────────────────────────
# COIN LİSTESİ (CoinGecko ID → Sembol)
# ──────────────────────────────────────────────────────────
COINS = {
"bitcoin": "BTC",
"ethereum": "ETH",
"binancecoin": "BNB",
"ripple": "XRP",
"solana": "SOL",
"cardano": "ADA",
"dogecoin": "DOGE",
"avalanche-2": "AVAX",
"chainlink": "LINK",
"polkadot": "DOT",
"matic-network": "MATIC",
"tron": "TRX",
"litecoin": "LTC",
"uniswap": "UNI",
"cosmos": "ATOM",
"near": "NEAR",
"algorand": "ALGO",
"stellar": "XLM",
"filecoin": "FIL",
"internet-computer": "ICP",
"aptos": "APT",
"arbitrum": "ARB",
"optimism": "OP",
"sui": "SUI",
"pepe": "PEPE",
"shiba-inu": "SHIB",
"the-sandbox": "SAND",
"decentraland": "MANA",
"aave": "AAVE",
"maker": "MKR",
}
BASE_URL = "https://api.coingecko.com/api/v3"
HEADERS = {"accept": "application/json"}
# ──────────────────────────────────────────────────────────
# VERİ ÇEKME
# ──────────────────────────────────────────────────────────
def fetch_ohlc(coin_id: str, days: int = 30) -> pd.Series | None:
"""CoinGecko OHLC endpoint – günlük kapanış fiyatları."""
url = f"{BASE_URL}/coins/{coin_id}/ohlc?vs_currency=usd&days={days}"
try:
r = requests.get(url, headers=HEADERS, timeout=20)
r.raise_for_status()
data = r.json()
if not data:
return None
df = pd.DataFrame(data, columns=["ts", "open", "high", "low", "close"])
df["ts"] = pd.to_datetime(df["ts"], unit="ms")
df = df.set_index("ts").resample("D").last().dropna()
return df["close"]
except Exception:
return None
def fetch_market_chart(coin_id: str, days: int = 30) -> pd.Series | None:
"""Fallback: /market_chart ile günlük fiyat."""
url = f"{BASE_URL}/coins/{coin_id}/market_chart?vs_currency=usd&days={days}&interval=daily"
try:
r = requests.get(url, headers=HEADERS, timeout=20)
r.raise_for_status()
data = r.json().get("prices", [])
if not data:
return None
df = pd.DataFrame(data, columns=["ts", "price"])
df["ts"] = pd.to_datetime(df["ts"], unit="ms")
df = df.set_index("ts").resample("D").last().dropna()
return df["price"]
except Exception:
return None
def get_price_series(coin_id: str, days: int) -> pd.Series | None:
s = fetch_ohlc(coin_id, days)
if s is None or len(s) < 5:
s = fetch_market_chart(coin_id, days)
return s if s is not None and len(s) >= 5 else None
# ──────────────────────────────────────────────────────────
# HESAPLAMALAR
# ──────────────────────────────────────────────────────────
def returns(series: pd.Series) -> pd.Series:
return series.pct_change().dropna()
def rolling_corr(s1: pd.Series, s2: pd.Series, window: int = 14) -> float:
"""Son N günlük rolling correlation."""
r1 = returns(s1)
r2 = returns(s2)
combined = pd.concat([r1, r2], axis=1).dropna()
if len(combined) < window:
return float(combined.iloc[:, 0].corr(combined.iloc[:, 1]))
tail = combined.tail(window)
return float(tail.iloc[:, 0].corr(tail.iloc[:, 1]))
def zscore_spread(s1: pd.Series, s2: pd.Series, window: int = 20) -> dict | None:
"""
Pairs trading Z-score.
ratio = log(s1) - log(s2)
z = (ratio - mean) / std (rolling window)
"""
log1 = np.log(s1)
log2 = np.log(s2)
ratio = (log1 - log2).dropna()
if len(ratio) < window:
window = max(5, len(ratio) // 2)
roll_mean = ratio.rolling(window).mean()
roll_std = ratio.rolling(window).std()
zscore = ((ratio - roll_mean) / roll_std).dropna()
if len(zscore) == 0:
return None
current_z = float(zscore.iloc[-1])
current_ratio = float(ratio.iloc[-1])
ratio_mean = float(ratio.mean())
ratio_std = float(ratio.std())
half_life = _half_life(ratio)
return {
"current_z": current_z,
"current_ratio": current_ratio,
"ratio_mean": ratio_mean,
"ratio_std": ratio_std,
"half_life": half_life,
}
def _half_life(spread: pd.Series) -> float:
"""Ornstein-Uhlenbeck half-life (OLS)."""
try:
lag = spread.shift(1).dropna()
delta = spread.diff().dropna()
combined = pd.concat([lag, delta], axis=1).dropna()
if len(combined) < 5:
return float("nan")
x = combined.iloc[:, 0].values
y = combined.iloc[:, 1].values
A = np.vstack([np.ones(len(x)), x]).T
result = np.linalg.lstsq(A, y, rcond=None)
b = result[0][1]
if b >= 0:
return float("nan")
return round(-np.log(2) / b, 1)
except Exception:
return float("nan")
def trade_signal(z: float, corr: float, half_life: float) -> tuple[str, str]:
"""Sinyal üret."""
if abs(z) < 0.5:
return "NÖTR", "#555577"
if corr < 0.5:
return "DÜŞÜK KOR.", "#443344"
hl_ok = not np.isnan(half_life) and 2 <= half_life <= 60
if abs(z) >= 2.5 and hl_ok:
strength = "GÜÇLÜ"
elif abs(z) >= 1.5:
strength = "ORTA"
else:
strength = "İZLE"
if z > 2.0:
return f"{strength} · 1.SAT / 2.AL", "red"
elif z < -2.0:
return f"{strength} · 1.AL / 2.SAT", "green"
elif z > 1.0:
return f"{strength} · 1.SAT / 2.AL", "orange"
else:
return f"{strength} · 1.AL / 2.SAT", "yellow"
# ──────────────────────────────────────────────────────────
# ANA TARAMA
# ──────────────────────────────────────────────────────────
def scan_correlations(
selected_coins: list,
days: int,
min_corr: float,
min_abs_z: float,
progress=gr.Progress(),
):
if not selected_coins or len(selected_coins) < 2:
return _err_html("En az 2 coin seçin!"), None, "Tarama yapılmadı."
coin_ids = [k for k, v in COINS.items() if v in selected_coins]
n = len(coin_ids)
# 1) Fiyat serilerini çek
price_data = {}
for i, cid in enumerate(coin_ids):
sym = COINS[cid]
progress((i + 1) / (n + 1), desc=f"{sym} fiyatları çekiliyor...")
s = get_price_series(cid, days)
if s is not None:
price_data[sym] = s
time.sleep(0.35)
available = list(price_data.keys())
if len(available) < 2:
return _err_html("Yeterli fiyat verisi alınamadı. Tekrar deneyin."), None, "Hata."
# 2) Tüm çiftleri analiz et
pairs = []
total_pairs = len(available) * (len(available) - 1) // 2
idx = 0
for i in range(len(available)):
for j in range(i + 1, len(available)):
idx += 1
progress(idx / max(total_pairs, 1), desc=f"{available[i]}-{available[j]} analiz...")
s1 = price_data[available[i]]
s2 = price_data[available[j]]
common = s1.index.intersection(s2.index)
if len(common) < 10:
continue
s1c = s1.loc[common]
s2c = s2.loc[common]
corr = rolling_corr(s1c, s2c, window=min(14, len(common) - 1))
if abs(corr) < min_corr:
continue
spread_info = zscore_spread(s1c, s2c, window=min(20, len(common) - 2))
if spread_info is None:
continue
z = spread_info["current_z"]
if abs(z) < min_abs_z:
continue
hl = spread_info["half_life"]
signal, _ = trade_signal(z, corr, hl)
pairs.append({
"Coin 1": available[i],
"Coin 2": available[j],
"Korelasyon": round(corr, 4),
"Z-Score": round(z, 3),
"Half-Life (gün)": round(hl, 1) if not np.isnan(hl) else "-",
"Ratio (log)": round(spread_info["current_ratio"], 5),
"Ratio Ort.": round(spread_info["ratio_mean"], 5),
"Sinyal": signal,
})
if not pairs:
return _empty_html(min_corr, min_abs_z), None, f"Uygun çift bulunamadı (corr≥{min_corr}, |z|≥{min_abs_z})."
df = (
pd.DataFrame(pairs)
.sort_values("Z-Score", key=abs, ascending=False)
.reset_index(drop=True)
)
html = _render_html(df)
log = _make_log(df, days, available)
return html, df, log
# ──────────────────────────────────────────────────────────
# HTML RENDER
# ──────────────────────────────────────────────────────────
def _signal_style(sig: str) -> str:
if "GÜÇLÜ" in sig and "SAT" in sig:
return "background:#ff6b6b22;color:#ff6b6b;border:1px solid #ff6b6b55"
if "GÜÇLÜ" in sig and "AL" in sig:
return "background:#00ff8822;color:#00ff88;border:1px solid #00ff8855"
if "ORTA" in sig and "SAT" in sig:
return "background:#ff9f4322;color:#ff9f43;border:1px solid #ff9f4355"
if "ORTA" in sig and "AL" in sig:
return "background:#ffd70022;color:#ffd700;border:1px solid #ffd70055"
if "İZLE" in sig:
return "background:#7b7bff22;color:#7b7bff;border:1px solid #7b7bff55"
return "color:#555;border:1px solid #333"
def _corr_bar(c: float) -> str:
pct = int(abs(c) * 100)
color = "#00d2ff" if c > 0 else "#ff6b9d"
return (
f""
)
def _z_badge(z: float) -> str:
az = abs(z)
if az >= 2.5:
c, sz = ("#ff4444" if z > 0 else "#00ff88"), "17px"
elif az >= 1.5:
c, sz = ("#ff9f43" if z > 0 else "#ffd700"), "15px"
else:
c, sz = "#7b7bff", "13px"
arrow = "▲" if z > 0 else "▼"
return f"{arrow} {z:+.3f}"
def _hl_badge(hl) -> str:
if hl == "-" or (isinstance(hl, float) and np.isnan(hl)):
return "—"
hl = float(hl)
if 2 <= hl <= 60:
c = "#00ff88"
elif hl < 2:
c = "#ff6b6b"
else:
c = "#555"
return f"{hl:.1f} gün"
def _render_html(df: pd.DataFrame) -> str:
rows = ""
for rank, (_, row) in enumerate(df.iterrows(), 1):
sig = row["Sinyal"]
sstyle = _signal_style(sig)
rows += f"""
| {rank} |
{row['Coin 1']}
⟷
{row['Coin 2']}
|
{_corr_bar(row['Korelasyon'])} |
{_z_badge(row['Z-Score'])} |
{_hl_badge(row['Half-Life (gün)'])} |
{sig}
|
"""
strong = len(df[df["Sinyal"].str.contains("GÜÇLÜ", na=False)])
orta = len(df[df["Sinyal"].str.contains("ORTA", na=False)])
izle = len(df[df["Sinyal"].str.contains("İZLE", na=False)])
return f"""
⚡ KORELASYon TİCARETİ · {datetime.now().strftime('%H:%M:%S')}
| # | Çift | Korelasyon |
Z-Score | Half-Life | Sinyal |
{rows}
"""
def _err_html(msg: str) -> str:
return (
f"⚠️ {msg}
"
)
def _empty_html(min_corr: float, min_abs_z: float) -> str:
return (
f""
f"
📭
"
f"
corr≥{min_corr} ve |Z|≥{min_abs_z} koşullarını sağlayan çift bulunamadı.
"
f"
Eşikleri düşürün veya daha fazla coin seçin.
KORELASYON TRADİNG SKANER
PAIRS TRADING · Z-SCORE · MEAN REVERSION · COINGECKO · API ANAHTARI YOK
"""
INFO_HTML = """
📐 ALGORİTMA
Z-Score = (log_ratio − ort) / std
Half-Life = OU modeli ile mean-reversion hızı
|Z| > 2 → İşlem sinyali
2 – 60 gün → İdeal half-life aralığı
Korelasyon > 0.7 → Güvenilir çift
"""
def export_csv(df):
if df is None:
return None
path = "/tmp/korelasyon_sonuclari.csv"
df.to_csv(path, index=False, encoding="utf-8-sig")
return path
with gr.Blocks(title="Korelasyon Trading Skaner") as demo:
gr.HTML(HEADER_HTML)
with gr.Row():
# ── Sol panel: coin seçimi ──
with gr.Column(scale=1, min_width=200):
coin_selector = gr.CheckboxGroup(
choices=ALL_SYMBOLS,
value=DEFAULT_COINS,
label="COİN SEÇİMİ",
)
gr.HTML(INFO_HTML)
# ── Sağ panel: kontroller + sonuç ──
with gr.Column(scale=3):
with gr.Row():
days_slider = gr.Slider(
minimum=14, maximum=90, value=30, step=7,
label="VERİ SÜRESİ (GÜN)",
)
corr_slider = gr.Slider(
minimum=0.3, maximum=0.99, value=0.65, step=0.05,
label="MİN. KORELASYON",
)
z_slider = gr.Slider(
minimum=0.5, maximum=4.0, value=1.5, step=0.25,
label="MİN. |Z-SCORE|",
)
with gr.Row():
scan_btn = gr.Button("⚡ TARAMAYI BAŞLAT", variant="primary", size="lg")
clear_btn = gr.Button("🗑 TEMİZLE", variant="secondary")
export_btn = gr.Button("📥 CSV İNDİR", variant="secondary")
result_html = gr.HTML(
value=(
"Coin seçip taramayı başlatın →
"
)
)
with gr.Accordion("📋 LOG / DETAY", open=False):
log_box = gr.Textbox(
label="",
lines=14,
max_lines=22,
placeholder="Tarama logları burada görünür...",
)
df_state = gr.State(None)
csv_file = gr.File(label="CSV Dosyası", visible=False)
# ── Bağlantılar ──
scan_btn.click(
fn=scan_correlations,
inputs=[coin_selector, days_slider, corr_slider, z_slider],
outputs=[result_html, df_state, log_box],
)
clear_btn.click(
fn=lambda: (
"Temizlendi.
",
None,
"",
),
outputs=[result_html, df_state, log_box],
)
export_btn.click(
fn=export_csv,
inputs=[df_state],
outputs=[csv_file],
).then(fn=lambda: gr.update(visible=True), outputs=[csv_file])
if __name__ == "__main__":
demo.launch(css=CSS)