""" Streamlit 의존이 없는 핵심 비즈니스 로직. - Binance Futures API 데이터 수집 - 지표 / 신호 계산 - 알림 (텔레그램) + 트레이드 lifecycle 기록 - 알림 / 일일 리포트 백그라운드 루프 기존 app_streamlit.py 에서 그대로 추출. FastAPI 에서도 그대로 import. """ import os import time import threading import requests import pandas as pd import numpy as np from datetime import datetime, timezone, timedelta import ta import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) import alert_state import settings_db import trades_db BASE = "https://fapi.binance.com" KST = timedelta(hours=9) STOP_LOSS_PCT = 0.0075 LONG_SIGNALS = {"strong_long_signal", "long_signal", "vol_long_signal", "reversal_long_signal"} SHORT_SIGNALS = {"strong_short_signal", "short_signal", "vol_short_signal", "reversal_short_signal"} TF_LABEL_MAP = { "1m": "1분봉", "3m": "3분봉", "5m": "5분봉", "15m": "15분봉", "30m": "30분봉", "1h": "1시간봉", "4h": "4시간봉", "12h": "12시간봉", "1d": "1일봉", "3d": "3일봉", "1M": "1개월봉", } SIG_DEFS = [ ("strong_long_signal", "strong_long", "🟢 강한 롱", "long"), ("strong_short_signal", "strong_short", "🔴 강한 숏", "short"), ("long_signal", "long", "🔼 일반 롱", "long"), ("short_signal", "short", "🔽 일반 숏", "short"), ("vol_long_signal", "vol_long", "🔼 볼륨 롱", "long"), ("vol_short_signal", "vol_short", "🔽 볼륨 숏", "short"), ("reversal_long_signal", "rev_long", "🔄 롱 추세 꺾임 감지", "long"), ("reversal_short_signal","rev_short", "🔄 숏 추세 꺾임 감지", "short"), ("short_caution_signal", "short_caution","⚠️ 숏 주의", "caution"), ] # ── 설정 helpers ── def TELEGRAM_TOKEN(): return settings_db.get("telegram_token", "") def TELEGRAM_CHAT_ID(): return settings_db.get("telegram_chat_id", "") def ALERT_COOLDOWN(): return settings_db.get_int("alert_cooldown_sec", 600) def STOP_LOSS_PCT_v(): return settings_db.get_float("stop_loss_pct", 0.0075) # ── 텔레그램 ── def send_telegram(message: str): token = TELEGRAM_TOKEN() chat_id = TELEGRAM_CHAT_ID() if not token or not chat_id: print("[텔레그램] 토큰/chat_id 미설정 — skip") return try: url = f"https://api.telegram.org/bot{token}/sendMessage" requests.post(url, data={"chat_id": chat_id, "text": message}, timeout=10) except Exception as e: print(f"[텔레그램 오류] {e}") # ── Binance Futures fetch ── def get_klines(symbol="BTCUSDT", interval="5m", limit=375): url = f"{BASE}/fapi/v1/klines" r = requests.get(url, params={"symbol": symbol, "interval": interval, "limit": limit}, timeout=10, verify=False) df = pd.DataFrame(r.json(), columns=[ "open_time","open","high","low","close","volume", "close_time","quote_vol","trades","taker_buy_vol","taker_sell_vol","ignore" ]) for c in ["open","high","low","close","volume","taker_buy_vol","taker_sell_vol"]: df[c] = df[c].astype(float) df["taker_sell_vol"] = df["volume"] - df["taker_buy_vol"] df["open_time"] = pd.to_datetime(df["open_time"], unit="ms") + KST return df def get_funding_rate(symbol="BTCUSDT", limit=100): url = f"{BASE}/fapi/v1/fundingRate" r = requests.get(url, params={"symbol": symbol, "limit": limit}, timeout=10, verify=False) df = pd.DataFrame(r.json()) if df.empty: return df df["fundingRate"] = df["fundingRate"].astype(float) * 100 df["fundingTime"] = pd.to_datetime(df["fundingTime"], unit="ms") + KST return df def get_open_interest_history(symbol="BTCUSDT", period="5m", limit=100): url = f"{BASE}/futures/data/openInterestHist" r = requests.get(url, params={"symbol": symbol, "period": period, "limit": limit}, timeout=10, verify=False) df = pd.DataFrame(r.json()) if df.empty: return df df["sumOpenInterest"] = df["sumOpenInterest"].astype(float) df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms") + KST return df def get_long_short_ratio(symbol="BTCUSDT", period="5m", limit=500): url = f"{BASE}/futures/data/topLongShortPositionRatio" r = requests.get(url, params={"symbol": symbol, "period": period, "limit": limit}, timeout=10, verify=False) df = pd.DataFrame(r.json()) if df.empty: return df df["longShortRatio"] = df["longShortRatio"].astype(float) df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms") + KST return df def _to_floor_freq(period): return {"1m":"1min","3m":"3min","5m":"5min","15m":"15min","30m":"30min","1h":"1h","4h":"4h","12h":"12h","1d":"1D","3d":"3D","1M":"1ME"}.get(period, period) # ── 지표 + 신호 ── def compute_indicators(df, interval="5m"): c = df["close"] df["MA7"] = c.rolling(7).mean() df["MA25"] = c.rolling(25).mean() df["MA99"] = c.rolling(99).mean() df["MA200"] = c.rolling(200).mean() df["BB_mid"] = c.rolling(20).mean() df["BB_std"] = c.rolling(20).std() df["BB_upper"] = df["BB_mid"] + 2 * df["BB_std"] df["BB_lower"] = df["BB_mid"] - 2 * df["BB_std"] df["RSI"] = ta.momentum.RSIIndicator(c, window=14).rsi() macd = ta.trend.MACD(c, window_slow=26, window_fast=12, window_sign=9) df["MACD"] = macd.macd() df["MACD_signal"] = macd.macd_signal() df["MACD_hist"] = macd.macd_diff() stoch = ta.momentum.StochRSIIndicator(c, window=14, smooth1=3, smooth2=3) df["StochRSI_k"] = stoch.stochrsi_k() * 100 df["StochRSI_d"] = stoch.stochrsi_d() * 100 df["ATR"] = ta.volatility.AverageTrueRange(df["high"], df["low"], df["close"], window=14).average_true_range() df = compute_signals(df, interval) return df def compute_signals(df, interval="5m"): LONG_RSI_MAX = settings_db.get_float("long_rsi_max", 75.0) SHORT_RSI_MIN = settings_db.get_float("short_rsi_min", 25.0) SLONG_RSI_MAX = settings_db.get_float("strong_long_rsi_max", 65.0) SSHORT_RSI_MIN = settings_db.get_float("strong_short_rsi_min", 35.0) BODY_PCT_MIN = settings_db.get_float("body_pct_min", 0.002) REV_BODY_PCT = settings_db.get_float("reversal_body_pct", 0.003) REV_VOL_MULT = settings_db.get_float("reversal_vol_mult", 1.3) VOL_EXH_MULT = settings_db.get_float("vol_exhaustion_mult", 3.0) VOL_NET_MULT = settings_db.get_float("vol_net_mult", 2.0) OI_ACTIVE_PCT = settings_db.get_float("oi_active_pct", 0.001) FR_SHORT_EXTREME = settings_db.get_float("fr_short_extreme", -0.007) df["bull_ma_2"] = (df["close"] > df["MA7"]) & (df["close"] > df["MA25"]) df["bear_ma_2"] = (df["close"] < df["MA7"]) & (df["close"] < df["MA25"]) df["bull_ma"] = (df["close"] > df["MA7"]) & (df["MA7"] > df["MA25"]) df["bear_ma"] = (df["close"] < df["MA7"]) & (df["MA7"] < df["MA25"]) bb_range = (df["BB_upper"] - df["BB_lower"]).replace(0, float("nan")) df["bb_pos"] = (df["close"] - df["BB_lower"]) / bb_range body_pct = (df["close"] - df["open"]) / df["open"].replace(0, float("nan")) df["long_signal"] = df["bull_ma_2"] & (df["RSI"] < LONG_RSI_MAX) & (df["MACD_hist"] > df["MACD_hist"].shift(1)) & (df["close"] > df["BB_mid"]) & (body_pct >= BODY_PCT_MIN) df["short_signal"] = df["bear_ma_2"] & (df["RSI"] > SHORT_RSI_MIN) & (df["MACD_hist"] < df["MACD_hist"].shift(1)) & (df["close"] < df["BB_mid"]) & (body_pct <= -BODY_PCT_MIN) df["long_signal"] = df["long_signal"] & (df["long_signal"].rolling(5, min_periods=1).sum().shift(1).fillna(0) == 0) df["short_signal"] = df["short_signal"] & (df["short_signal"].rolling(5, min_periods=1).sum().shift(1).fillna(0) == 0) if "sumOpenInterest" in df.columns and df["sumOpenInterest"].notna().sum() > 5: oi_series = df["sumOpenInterest"].ffill() else: oi_series = df["close"] * df["volume"] df["oi_up"] = oi_series > oi_series.shift(1) df["oi_down"] = oi_series < oi_series.shift(1) df["oi_up_2"] = df["oi_up"] & df["oi_up"].shift(1).fillna(False) df["oi_down_2"] = df["oi_down"] & df["oi_down"].shift(1).fillna(False) df["oi_active"] = oi_series.pct_change().abs() > OI_ACTIVE_PCT df["taker_buy_dom"] = df["taker_buy_vol"] > df["taker_sell_vol"] df["taker_sell_dom"] = df["taker_sell_vol"] > df["taker_buy_vol"] df["taker_buy_2"] = df["taker_buy_dom"] & df["taker_buy_dom"].shift(1).fillna(False) df["taker_sell_2"] = df["taker_sell_dom"] & df["taker_sell_dom"].shift(1).fillna(False) df["fr_long_favor"] = df["taker_buy_vol"].rolling(3).mean() > df["taker_sell_vol"].rolling(3).mean() df["fr_short_favor"] = df["taker_sell_vol"].rolling(3).mean() > df["taker_buy_vol"].rolling(3).mean() df["strong_long_signal"] = df["bull_ma_2"] & (df["RSI"] < SLONG_RSI_MAX) & (df["MACD_hist"] > df["MACD_hist"].shift(1)) & df["oi_up_2"] & df["taker_buy_2"] & df["fr_long_favor"] & (df["close"] > df["open"]) df["strong_short_signal"] = df["bear_ma_2"] & (df["RSI"] > SSHORT_RSI_MIN) & (df["MACD_hist"] < df["MACD_hist"].shift(1)) & df["oi_down_2"] & df["taker_sell_2"] & df["fr_short_favor"] & (df["close"] < df["open"]) df["strong_long_signal"] = df["strong_long_signal"] & (df["strong_long_signal"].rolling(10, min_periods=1).sum().shift(1).fillna(0) == 0) df["strong_short_signal"] = df["strong_short_signal"] & (df["strong_short_signal"].rolling(10, min_periods=1).sum().shift(1).fillna(0) == 0) vol_avg = df["volume"].rolling(10).mean() spike = df["volume"] > vol_avg * VOL_EXH_MULT buy_spike = spike & (df["taker_buy_vol"] > df["taker_sell_vol"]) sell_spike = spike & (df["taker_sell_vol"] > df["taker_buy_vol"]) df["exhaustion_short"] = buy_spike.shift(1).fillna(False) df["exhaustion_long"] = sell_spike.shift(1).fillna(False) _vol_min_map = {"1m": 33, "3m": 100, "5m": 100, "15m": 300, "30m": 600, "1h": 1200, "2h": 2400, "4h": 4800, "12h": 14400, "1d": 28800, "3d": 86400, "1M": 864000} _vol_min = _vol_min_map.get(interval, 100) df["sell_net"] = df["taker_sell_vol"] - df["taker_buy_vol"] sell_net_avg = df["sell_net"].rolling(10).mean() sell_spike_strong = ( (df["sell_net"] > sell_net_avg * VOL_NET_MULT) & (df["sell_net"] > 0) & (df["taker_sell_vol"] > _vol_min) & df["oi_active"] ) cooldown_vol_short = sell_spike_strong.rolling(10, min_periods=1).sum().shift(1).fillna(0) == 0 df["vol_short_signal"] = sell_spike_strong & cooldown_vol_short df["buy_net"] = df["taker_buy_vol"] - df["taker_sell_vol"] buy_net_avg = df["buy_net"].rolling(10).mean() buy_spike_strong = ( (df["buy_net"] > buy_net_avg * VOL_NET_MULT) & (df["buy_net"] > 0) & (df["taker_buy_vol"] > _vol_min) & df["oi_active"] ) cooldown_vol_long = buy_spike_strong.rolling(10, min_periods=1).sum().shift(1).fillna(0) == 0 df["vol_long_signal"] = buy_spike_strong & cooldown_vol_long if "fundingRate" in df.columns and "sumOpenInterest" in df.columns: fr_extreme = df["fundingRate"] <= FR_SHORT_EXTREME raw_signal = df["oi_down_2"] & fr_extreme cooldown_mask = raw_signal.rolling(5, min_periods=1).sum().shift(1).fillna(0) == 0 df["short_caution_signal"] = raw_signal & cooldown_mask else: df["short_caution_signal"] = False prior_close = df["close"].shift(1) prior_close_3 = df["close"].shift(3) was_up = prior_close > prior_close_3 was_down = prior_close < prior_close_3 candle_body_pct = (df["close"] - df["open"]) / df["open"].replace(0, float("nan")) vol_avg3 = df["volume"].rolling(3).mean().shift(1) vol_strong = df["volume"] > vol_avg3 * REV_VOL_MULT rev_short_raw = was_up & (candle_body_pct < -REV_BODY_PCT) & vol_strong rev_long_raw = was_down & (candle_body_pct > REV_BODY_PCT) & vol_strong df["reversal_short_signal"] = rev_short_raw & (rev_short_raw.rolling(3, min_periods=1).sum().shift(1).fillna(0) == 0) df["reversal_long_signal"] = rev_long_raw & (rev_long_raw.rolling(3, min_periods=1).sum().shift(1).fillna(0) == 0) return df def build_signal_df(symbol, interval, klines_limit=200): """알림 / API 공용 - klines + OI + FR 머지 + 지표/신호 계산""" df = get_klines(symbol, interval, klines_limit) oi_period = interval if interval in ["5m","15m","30m","1h","4h","12h","1d","3d","1M"] else "5m" try: oi = get_open_interest_history(symbol, oi_period, min(klines_limit, 500)) if not oi.empty: oi_m = oi[["timestamp","sumOpenInterest"]].rename(columns={"timestamp":"open_time"}) df["open_time_r"] = df["open_time"].dt.floor(_to_floor_freq(oi_period)) oi_m["open_time"] = oi_m["open_time"].dt.floor(_to_floor_freq(oi_period)) df = df.merge(oi_m, left_on="open_time_r", right_on="open_time", how="left", suffixes=("","_oi")) df = df.drop(columns=["open_time_r","open_time_oi"], errors="ignore") df["sumOpenInterest"] = df["sumOpenInterest"].ffill() except Exception: pass try: fr = get_funding_rate(symbol, 200) if not fr.empty: fr_m = fr[["fundingTime","fundingRate"]].rename(columns={"fundingTime":"open_time"}) fr_m["open_time"] = fr_m["open_time"].dt.floor(_to_floor_freq("1h")) df["open_time_r2"] = df["open_time"].dt.floor(_to_floor_freq("1h")) df = df.merge(fr_m, left_on="open_time_r2", right_on="open_time", how="left", suffixes=("","_fr")) df = df.drop(columns=["open_time_r2","open_time_fr"], errors="ignore") df["fundingRate"] = df["fundingRate"].ffill().fillna(0) except Exception: pass df = compute_indicators(df, interval) return df # ── 알림 코어 ── def check_and_alert(df, symbol, interval): now = time.time() if df is None or df.empty: return forming_ct = df.iloc[-1]["open_time"] if interval not in alert_state.synced_intervals: for sig, key, _, _ in SIG_DEFS: if sig not in df.columns: continue triggered = df[df[sig].fillna(False)] if not triggered.empty: alert_state.last_fired_candle[(interval, key)] = triggered.iloc[-1]["open_time"] alert_state.synced_intervals.add(interval) print(f"[알림스레드] {interval} 초기 sync 완료") return new_pending = [] for p in alert_state.pending_groups: if p["interval"] != interval: new_pending.append(p) continue ct = p["candle_time"] row_match = df[df["open_time"] == ct] if row_match.empty: continue row = row_match.iloc[0] any_still_true = any(bool(row.get(s, False)) for s in p["sig_cols"]) if any_still_true: if ct == forming_ct: new_pending.append(p) else: send_telegram(f"[취소 알림]\n{p['msg']}") le = alert_state.long_entry.get(interval) se = alert_state.short_entry.get(interval) if p["direction"] == "long" and le is not None and le.get("open_time") == ct: trades_db.record_exit(symbol, interval, "long", ct, float(row["close"]), "cancelled") alert_state.long_entry[interval] = None elif p["direction"] == "short" and se is not None and se.get("open_time") == ct: trades_db.record_exit(symbol, interval, "short", ct, float(row["close"]), "cancelled") alert_state.short_entry[interval] = None alert_state.pending_groups = new_pending recent = df.tail(3) eligible = [] for sig, key, sub_label, direction in SIG_DEFS: if sig not in recent.columns: continue triggered = recent[recent[sig].fillna(False)] seen_key = (interval, sig) prev_seen = alert_state.signal_seen_count.get(seen_key) if triggered.empty: if prev_seen: alert_state.signal_seen_count[seen_key] = {"candle_time": prev_seen["candle_time"], "count": 0} continue candle_time = triggered.iloc[-1]["open_time"] state_key = (interval, key) if candle_time == alert_state.last_fired_candle.get(state_key): continue if now - alert_state.last_alert.get(state_key, 0) <= ALERT_COOLDOWN(): continue if prev_seen is None or prev_seen["candle_time"] != candle_time: alert_state.signal_seen_count[seen_key] = {"candle_time": candle_time, "count": 1} else: alert_state.signal_seen_count[seen_key] = {"candle_time": candle_time, "count": prev_seen["count"] + 1} count = alert_state.signal_seen_count[seen_key]["count"] stable_min = settings_db.get_int("forming_stable_polls", 2) if candle_time == forming_ct and count < stable_min: continue eligible.append({ "sig": sig, "key": key, "sub_label": sub_label, "direction": direction, "candle_time": candle_time, "row": triggered.iloc[-1], }) groups = {"long": [], "short": [], "caution": []} for e in eligible: groups[e["direction"]].append(e) tf_label = TF_LABEL_MAP.get(interval, interval) def _send_group(group): if not group: return candle_time = group[0]["candle_time"] candle_time_str = pd.Timestamp(candle_time).strftime("%Y-%m-%d %H:%M") sub_labels = " + ".join(e["sub_label"] for e in group) direction = group[0]["direction"] trades_db.log_signal_events(symbol, interval, group) if direction == "caution": msg = f"{sub_labels} 신호\n{symbol} {tf_label}\n시간: {candle_time_str}" send_telegram(msg) else: entry_price = float(group[0]["row"]["open"]) sl_pct = STOP_LOSS_PCT_v() stop_price = entry_price * (1 - sl_pct) if direction == "long" else entry_price * (1 + sl_pct) msg = ( f"{sub_labels} 진입 신호\n{symbol} {tf_label}\n" f"시간: {candle_time_str}\n진입가: {entry_price:,.2f}\n손절가: {stop_price:,.2f}" ) entry_record = {"price": entry_price, "stop": stop_price, "open_time": candle_time, "entry_msg": msg} if interval in ("30m", "1h"): opposite_dict = alert_state.short_entry if direction == "long" else alert_state.long_entry opposite_label = "숏" if direction == "long" else "롱" opposite_direction = "short" if direction == "long" else "long" for opp_interval, opp_rec in list(opposite_dict.items()): if opp_rec is None: continue send_telegram( f"[반대 신호 감지 - {opposite_label} 청산 권장]\n" f"--- 기존 진입 ---\n{opp_rec['entry_msg']}\n" f"--- 반대 신호 ---\n{msg}" ) trades_db.record_exit(symbol, opp_interval, opposite_direction, opp_rec.get("open_time"), entry_price, "reversal") opposite_dict[opp_interval] = None if direction == "long": alert_state.long_entry[interval] = entry_record else: alert_state.short_entry[interval] = entry_record trades_db.record_entry(symbol, interval, direction, [e["sig"] for e in group], candle_time, entry_price, stop_price) send_telegram(msg) for e in group: alert_state.last_alert[(interval, e["key"])] = now alert_state.last_fired_candle[(interval, e["key"])] = e["candle_time"] if candle_time == forming_ct: alert_state.pending_groups.append({ "interval": interval, "direction": direction, "candle_time": candle_time, "msg": msg, "sig_cols": [e["sig"] for e in group], }) _send_group(groups.get("long", [])) _send_group(groups.get("short", [])) _send_group(groups.get("caution", [])) current_price = float(df.iloc[-1]["close"]) le = alert_state.long_entry.get(interval) se = alert_state.short_entry.get(interval) if le is not None and current_price <= le["stop"]: send_telegram(f"[손절가알림]\n{le['entry_msg']}\n현재가: {current_price:,.2f}") trades_db.record_exit(symbol, interval, "long", le.get("open_time"), current_price, "stop_loss") alert_state.long_entry[interval] = None if se is not None and current_price >= se["stop"]: send_telegram(f"[손절가알림]\n{se['entry_msg']}\n현재가: {current_price:,.2f}") trades_db.record_exit(symbol, interval, "short", se.get("open_time"), current_price, "stop_loss") alert_state.short_entry[interval] = None # ── 백그라운드 루프 (FastAPI startup 에서 호출) ── def alert_timeframes(): return settings_db.get_list("alert_timeframes", default=["5m", "15m", "30m", "1h"]) def alert_loop(): while True: poll = max(10, settings_db.get_int("polling_interval_sec", 30)) if not settings_db.get_bool("alert_enabled", True): time.sleep(poll) continue with alert_state.alert_lock: symbol = alert_state.alert_symbol for interval in alert_timeframes(): try: df = build_signal_df(symbol, interval, 200) check_and_alert(df, symbol, interval) except Exception as e: print(f"[알림스레드 오류] {interval}: {e}") time.sleep(poll) def daily_report_loop(): while True: try: if not settings_db.get_bool("daily_report_enabled", True): time.sleep(60) continue now_kst = (datetime.now(timezone.utc) + KST).replace(tzinfo=None) today_str = now_kst.strftime("%Y-%m-%d") if alert_state.last_report_date is None: alert_state.last_report_date = today_str print(f"[일일리포트] 스레드 기동 -- 다음 자정({today_str} 24:00 KST) 까지 대기") elif alert_state.last_report_date != today_str: print(f"[일일리포트] 자정 통과 감지 -> 발송 ({today_str})") # send_daily_report 는 app_streamlit.py 안에 있는 그대로 사용 (없어도 silent skip) try: from app_streamlit import send_daily_report with alert_state.alert_lock: symbol = alert_state.alert_symbol send_daily_report(symbol) except Exception as e: print(f"[일일리포트 호출 실패] {e}") alert_state.last_report_date = today_str except Exception as e: print(f"[일일리포트 스레드 오류] {e}") time.sleep(60) def start_background_threads(): """FastAPI startup 에서 호출. 한 번만 시작.""" if not alert_state.alert_started: t = threading.Thread(target=alert_loop, daemon=True, name="alert_loop") t.start() alert_state.alert_started = True if not alert_state.daily_report_started: dr = threading.Thread(target=daily_report_loop, daemon=True, name="daily_report_loop") dr.start() alert_state.daily_report_started = True