6e0c4508fa
- tick_trader.py를 Controller로 축소, 로직을 3개 모듈로 분리: - core/signal.py: 시그널 감지, 지표 계산 (calc_vr, calc_atr, detect_signal) - core/order.py: Upbit 주문 실행 (매수/매도/취소/조회) - core/position_manager.py: 포지션 관리, DB sync, 복구, 청산 조건 - type hints, Google docstring, 구체적 예외 타입 적용 - 50줄 초과 함수 분리 (process_signal, restore_positions) - 미사용 파일 58개 archive/ 폴더로 이동 - README.md 추가 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
82 lines
2.5 KiB
Python
82 lines
2.5 KiB
Python
"""10분봉 캐시 갱신 스크립트 — 최신 45일 데이터를 Upbit API로 재수집."""
|
|
|
|
import os, sys, pickle, time
|
|
from pathlib import Path
|
|
from datetime import datetime, timedelta
|
|
|
|
import pyupbit
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv(dotenv_path=Path(__file__).parent.parent / ".env")
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
CACHE_FILE = Path(__file__).parent.parent / "data" / "sim10m_cache.pkl"
|
|
TOP30_FILE = Path(__file__).parent.parent / "data" / "top30_tickers.pkl"
|
|
SIM_DAYS = 45
|
|
TOP_N = 20
|
|
|
|
|
|
def fetch_10m(ticker: str, days: int) -> "pd.DataFrame | None":
|
|
import pandas as pd
|
|
target_start = datetime.now() - timedelta(days=days)
|
|
all_dfs, to, prev_oldest = [], None, None
|
|
while True:
|
|
kwargs = dict(ticker=ticker, interval="minute10", count=200)
|
|
if to:
|
|
kwargs["to"] = to.strftime("%Y-%m-%d %H:%M:%S")
|
|
try:
|
|
df = pyupbit.get_ohlcv(**kwargs)
|
|
except Exception:
|
|
time.sleep(0.5)
|
|
break
|
|
if df is None or df.empty:
|
|
break
|
|
all_dfs.append(df)
|
|
oldest = df.index[0]
|
|
if prev_oldest is not None and oldest >= prev_oldest:
|
|
break
|
|
prev_oldest = oldest
|
|
if oldest <= target_start:
|
|
break
|
|
to = oldest
|
|
time.sleep(0.12)
|
|
if not all_dfs:
|
|
return None
|
|
combined = pd.concat(all_dfs).sort_index()
|
|
combined = combined[~combined.index.duplicated(keep="last")]
|
|
return combined[combined.index >= target_start]
|
|
|
|
|
|
def main():
|
|
# 현재 Top20 종목 가져오기
|
|
from core.market import get_top_tickers
|
|
print("Top20 종목 조회...")
|
|
tickers = get_top_tickers()[:TOP_N]
|
|
print(f" {tickers}\n")
|
|
|
|
data = {"10m": {}}
|
|
for i, ticker in enumerate(tickers, 1):
|
|
print(f"\r {i:>2}/{len(tickers)} {ticker} ", end="", flush=True)
|
|
df = fetch_10m(ticker, SIM_DAYS)
|
|
if df is not None and len(df) > 100:
|
|
data["10m"][ticker] = df
|
|
time.sleep(0.15)
|
|
|
|
print(f"\n\n종목: {len(data['10m'])}개")
|
|
if data["10m"]:
|
|
sample = next(iter(data["10m"].values()))
|
|
print(f"기간: {sample.index[0].strftime('%Y-%m-%d')} ~ {sample.index[-1].strftime('%Y-%m-%d')}")
|
|
print(f"레코드: {len(sample)}개")
|
|
|
|
# 저장
|
|
pickle.dump(data, open(CACHE_FILE, "wb"))
|
|
print(f"\n캐시 저장: {CACHE_FILE}")
|
|
|
|
# top30 갱신
|
|
pickle.dump(tickers, open(TOP30_FILE, "wb"))
|
|
print(f"종목 저장: {TOP30_FILE}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|