from __future__ import annotations
import json
import math
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from urllib.parse import urlencode
from urllib.request import Request, urlopen
import streamlit as st
import pandas as pd
# إعدادات صفحة الويب
st.set_page_config(
page_title="ماسح زخم Binance Spot",
page_icon="📈",
layout="wide",
)
BINANCE_API = "https://api.binance.com"
KLINE_INTERVAL = "5m"
KLINE_LIMIT = 12
CONCURRENCY = 32
RETRIES = 3
REQUEST_TIMEOUT_SECONDS = 12
CANDLE_DURATION_MS = 5 * 60 * 1000
SIGNAL_MAX_AGE_MS = CANDLE_DURATION_MS + 60 * 1000
@dataclass(frozen=True)
class Candle:
open_time: int
open: float
high: float
close: float
volume: float
close_time: int
@dataclass(frozen=True)
class Signal:
symbol: str
price: float
candle_open_time: int
candle_close_time: int
candle_age_seconds: int
candle_closed: bool
volume_multiple: float
body_multiple: float
def fetch_json(url: str):
for attempt in range(RETRIES):
request = Request(
url,
headers={
"Accept": "application/json",
"User-Agent": "Mozilla/5.0 momentum-scanner",
},
)
try:
with urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
return json.loads(response.read().decode("utf-8"))
except Exception:
if attempt == RETRIES - 1:
raise
time.sleep(0.9 * (attempt + 1))
raise RuntimeError("تعذّر جلب بيانات Binance")
def parse_candle(values: list) -> Candle:
return Candle(
open_time=int(values[0]),
open=float(values[1]),
high=float(values[2]),
close=float(values[4]),
volume=float(values[5]),
close_time=int(values[6]),
)
def is_quote_pair(item: dict, quote: str) -> bool:
symbol = item.get("symbol", "")
if not isinstance(symbol, str) or not symbol.endswith(quote):
return False
try:
return math.isfinite(float(item.get("price", "nan")))
except (TypeError, ValueError):
return False
def find_fresh_signal(raw_candles: list[list], now_ms: int) -> tuple | None:
candles = sorted(
(parse_candle(item) for item in raw_candles),
key=lambda candle: candle.open_time,
)
if len(candles) < 11:
return None
first_candidate = max(10, len(candles) - 2)
for index in range(len(candles) - 1, first_candidate - 1, -1):
candle = candles[index]
age_ms = now_ms - candle.open_time
if age_ms < 0 or age_ms > SIGNAL_MAX_AGE_MS:
continue
candle_closed = candle.close_time < now_ms
if not candle_closed and index != len(candles) - 1:
continue
if candle.close <= candle.open:
continue
previous_ten = candles[index - 10:index]
if len(previous_ten) != 10:
continue
if any(previous.close_time >= candle.open_time for previous in previous_ten):
continue
average_volume = sum(c.volume for c in previous_ten) / 10
average_body = sum(abs(c.close - c.open) for c in previous_ten) / 10
body = abs(candle.close - candle.open)
if average_volume <= 0 or average_body <= 0:
continue
volume_multiple = candle.volume / average_volume
body_multiple = body / average_body
if volume_multiple < 2 or body_multiple < 2:
continue
return (
candle,
max(0, age_ms // 1000),
candle_closed,
volume_multiple,
body_multiple,
)
return None
def scan_symbol(symbol: str) -> Signal | None:
query = urlencode({"symbol": symbol, "interval": KLINE_INTERVAL, "limit": KLINE_LIMIT})
raw_candles = fetch_json(f"{BINANCE_API}/api/v3/klines?{query}")
now_ms = int(time.time() * 1000)
match = find_fresh_signal(raw_candles, now_ms)
if match is None:
return None
candle, age_seconds, candle_closed, volume_multiple, body_multiple = match
return Signal(
symbol=symbol,
price=candle.close,
candle_open_time=candle.open_time,
candle_close_time=candle.close_time,
candle_age_seconds=age_seconds,
candle_closed=candle_closed,
volume_multiple=volume_multiple,
body_multiple=body_multiple,
)
@st.cache_data(ttl=30)
def run_scan(quote: str) -> list[Signal]:
normalized_quote = quote.strip().upper()
tickers = fetch_json(f"{BINANCE_API}/api/v3/ticker/price")
symbols = [item["symbol"] for item in tickers if is_quote_pair(item, normalized_quote)]
if not symbols:
return []
rows: list[Signal] = []
with ThreadPoolExecutor(max_workers=min(CONCURRENCY, len(symbols))) as pool:
futures = {pool.submit(scan_symbol, symbol): symbol for symbol in symbols}
for future in as_completed(futures):
try:
signal = future.result()
if signal is not None:
rows.append(signal)
except Exception:
pass
rows.sort(key=lambda r: (r.candle_age_seconds, -r.volume_multiple))
return rows
# --- الواجهة (UI) ---
st.title("🚀 ماسح زخم Binance Spot (فريم 5 دقائق)")
st.markdown("لوحة تحكم حية لمراقبة أزواج العملات التي تسجل طفرات في الحجم والجسم مقارنة بآخر 10 شموع.")
with st.sidebar:
st.header("إعدادات الفحص")
quote_currency = st.selectbox("رمز التسعير (Quote)", ["USDT", "USDC", "BTC"], index=0)
scan_button = st.button("بدء الفحص الآن 🔄", use_container_width=True)
if scan_button or "initialized" not in st.session_state:
st.session_state["initialized"] = True
with st.spinner(f"جاري فحص السوق لعملات {quote_currency}... يرجى الانتظار"):
st.session_state["signals"] = run_scan(quote_currency)
signals = st.session_state.get("signals", [])
col1, col2 = st.columns(2)
with col1:
st.metric("الإشارات المكتشفة", len(signals))
with col2:
st.metric("عملة التسعير", quote_currency)
st.divider()
if not signals:
st.warning("لا توجد إشارات زخم مطابقة للشروط حالياً.")
else:
data = []
for row in signals:
state = "أغلقت للتو 🟢" if row.candle_closed else "جارية الآن ⏳"
age_str = f"{row.candle_age_seconds // 60}:{row.candle_age_seconds % 60:02d}"
data.append({
"زوج التداول": row.symbol,
"السعر": f"{row.price:.8g}",
"حالة الشمعة": state,
"عمر الشمعة": age_str,
"مضاعف الحجم": f"{row.volume_multiple:.2f}×",
"مضاعف الجسم": f"{row.body_multiple:.2f}×",
})
df = pd.DataFrame(data)
st.dataframe(df, use_container_width=True, hide_index=True)