130 lines
4.8 KiB
Python
130 lines
4.8 KiB
Python
import math
|
||
import requests
|
||
|
||
import pandas as pd
|
||
import yfinance as yf
|
||
import matplotlib.pyplot as plt
|
||
plt.rcParams['font.family'] ='AppleGothic'
|
||
plt.rcParams['axes.unicode_minus'] =False
|
||
|
||
from config import *
|
||
from stock_monitor import calculate_technical_indicators, detect_turnaround_signal
|
||
|
||
# 비트/알트코인 KRW 마켓 식별: 문자열 "-KRW" 포함 여부로 간단 구분
|
||
|
||
INTERVAL_MAP = {
|
||
60: "60m", # 1시간 (yfinance)
|
||
240: "4h", # 4시간 (yfinance)
|
||
}
|
||
|
||
BITHUMB_MAX_COUNT = 3000 # API 최대 3000 캔들
|
||
|
||
|
||
def fetch_coin_history_bithumb(symbol: str, interval_minutes: int, days: int) -> pd.DataFrame:
|
||
"""빗썸 API를 이용해 최근 `days`일 코인 데이터 수집 (interval 60 / 240)"""
|
||
if interval_minutes not in (60, 240):
|
||
raise ValueError("Bithumb API only supports 60 or 240 minutes in this helper")
|
||
|
||
minutes = interval_minutes
|
||
count = int(math.ceil(days * 24 * 60 / minutes)) + 10 # 여유분 10 캔들
|
||
count = min(count, BITHUMB_MAX_COUNT)
|
||
|
||
url = f"https://api.bithumb.com/v1/candles/minutes/{minutes}?market=KRW-{symbol}&count={count}"
|
||
res = requests.get(url, timeout=5)
|
||
res.raise_for_status()
|
||
raw = res.json()
|
||
|
||
if not isinstance(raw, list) or len(raw) == 0:
|
||
raise RuntimeError("Empty response from Bithumb API")
|
||
|
||
df_temp = pd.DataFrame(raw)
|
||
# API 반환: [timestamp, open, close, high, low, volume] 순
|
||
df_temp = df_temp.sort_index(ascending=False) # 최신순, 뒤집어서 역순 전달
|
||
|
||
data = pd.DataFrame()
|
||
# data.columns = ['datetime', 'open', 'close', 'high', 'low', 'volume']
|
||
# data['datetime'] = pd.to_datetime(data_temp['candle_date_time_kst'])
|
||
data['datetime'] = pd.to_datetime(df_temp['candle_date_time_kst'], format='%Y-%m-%dT%H:%M:%S')
|
||
data['Open'] = df_temp['opening_price']
|
||
data['Close'] = df_temp['trade_price']
|
||
data['High'] = df_temp['high_price']
|
||
data['Low'] = df_temp['low_price']
|
||
data['Volume'] = df_temp['candle_acc_trade_volume']
|
||
data = data.set_index('datetime')
|
||
data = data.astype(float)
|
||
data["datetime"] = data.index
|
||
|
||
data = data.set_index("datetime").sort_index() # 시간 오름차순
|
||
return data
|
||
|
||
|
||
def fetch_price_history(symbol: str, interval_minutes: int, days: int = 7) -> pd.DataFrame:
|
||
"""최근 `days`일 데이터(캔들)를 가져온다. 코인(-KRW)은 빗썸, 그 외 yfinance."""
|
||
if symbol in KR_COINS:
|
||
base_symbol = symbol.replace("-KRW", "")
|
||
return fetch_coin_history_bithumb(base_symbol, interval_minutes, days)
|
||
|
||
# -------- 주식/ETF/해외코인 (yfinance) --------
|
||
if interval_minutes not in INTERVAL_MAP:
|
||
raise ValueError("interval must be 60 or 240")
|
||
|
||
interval_str = INTERVAL_MAP[interval_minutes]
|
||
|
||
df = yf.download(
|
||
tickers=symbol,
|
||
period=f"{days}d",
|
||
interval=interval_str,
|
||
progress=False,
|
||
)
|
||
|
||
if df.empty:
|
||
raise RuntimeError("No data fetched. Check symbol or interval support.")
|
||
|
||
return df
|
||
|
||
|
||
def run_simulation(symbol: str, interval_minutes: int, days: int = 7):
|
||
data = fetch_price_history(symbol, interval_minutes, days)
|
||
data = calculate_technical_indicators(data)
|
||
|
||
alerts = [] # (timestamp, price)
|
||
# 시계열 순회하며 알림 조건 체크
|
||
for i in range(len(data)):
|
||
slice_df = data.iloc[: i + 1]
|
||
info = detect_turnaround_signal(symbol, slice_df, interval=interval_minutes)
|
||
if info and info["alert"]:
|
||
alerts.append((slice_df.index[-1], slice_df["Close"].iloc[-1]))
|
||
|
||
# 8월 3일 이후의 매수 신호만 고려
|
||
alerts = [(time, price) for time, price in alerts if time > pd.Timestamp('2025-08-03')]
|
||
|
||
# Plot
|
||
plt.figure(figsize=(12, 6))
|
||
plt.plot(data.index, data["Close"], label="종가", color="black")
|
||
plt.plot(data.index, data["MA5"], label="MA5", color="orange", linewidth=1)
|
||
plt.plot(data.index, data["MA20"], label="MA20", color="blue", linewidth=1)
|
||
plt.plot(data.index, data["MA40"], label="MA40", color="green", linewidth=1)
|
||
# Bollinger Bands
|
||
plt.plot(data.index, data["Upper"], label="볼린저 Upper", color="grey", linestyle="--", linewidth=1)
|
||
plt.plot(data.index, data["Lower"], label="볼린저 Lower", color="grey", linestyle="--", linewidth=1)
|
||
plt.fill_between(data.index, data["Lower"], data["Upper"], color="grey", alpha=0.1)
|
||
|
||
if alerts:
|
||
times, prices = zip(*alerts)
|
||
plt.scatter(times, prices, facecolors='none', edgecolors='red', linewidths=2, s=150, zorder=6, label='매수신호')
|
||
|
||
plt.title(f"{symbol} – 시뮬레이션 {interval_minutes}분봉 (최근 {days}일)")
|
||
plt.xlabel("날짜")
|
||
plt.ylabel("가격")
|
||
plt.legend()
|
||
plt.grid(True)
|
||
plt.tight_layout()
|
||
plt.show()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
symbol = 'WLD'
|
||
interval = 240
|
||
days = 7
|
||
run_simulation(symbol, interval, days)
|