7f1921441b
- llm_advisor: Anthropic → OpenRouter API 전환 (claude-haiku-4.5) - llm_advisor: get_ticker_context DB tool 추가 (24h/7d 가격, 뉴스) - llm_advisor: 구조화 JSON 응답 (confidence, reason, market_status, watch_needed) - llm_advisor: LLM primary + cascade fallback (llm_active 플래그) - llm_advisor: SQL bind variable 버그 수정 (INTERVAL → NUMTODSINTERVAL) - tick_collector: backtest_ohlcv 1분봉 실시간 갱신 추가 (60초 주기) - context_collector: 신규 데몬 — 1시간마다 price_stats + SearXNG 뉴스 수집 - ecosystem: tick-collector, tick-trader, context-collector PM2 등록 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
227 lines
7.8 KiB
Python
227 lines
7.8 KiB
Python
"""종목 컨텍스트 수집 데몬.
|
|
|
|
1시간마다 각 종목의 평판 정보를 수집해 Oracle ticker_context 테이블에 저장.
|
|
LLM 어드바이저가 매도 판단 시 참조.
|
|
|
|
수집 항목:
|
|
- price_stats: 24h/7d 가격 변동률, 거래량 추이
|
|
- news: SearXNG 웹 검색 뉴스 요약
|
|
|
|
실행:
|
|
.venv/bin/python3 daemons/context_collector.py
|
|
로그:
|
|
/tmp/context_collector.log
|
|
"""
|
|
import sys, os, time, logging, json, requests
|
|
from datetime import datetime
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
from dotenv import load_dotenv
|
|
load_dotenv(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), '.env'))
|
|
|
|
import pyupbit
|
|
import oracledb
|
|
|
|
TICKERS = [
|
|
'KRW-XRP', 'KRW-BTC', 'KRW-ETH', 'KRW-SOL', 'KRW-DOGE',
|
|
'KRW-ADA', 'KRW-SUI', 'KRW-NEAR', 'KRW-KAVA', 'KRW-SXP',
|
|
'KRW-AKT', 'KRW-SONIC', 'KRW-IP', 'KRW-ORBS', 'KRW-VIRTUAL',
|
|
'KRW-BARD', 'KRW-XPL', 'KRW-KITE', 'KRW-ENSO', 'KRW-0G',
|
|
'KRW-MANTRA', 'KRW-EDGE', 'KRW-CFG', 'KRW-ARDR', 'KRW-SIGN',
|
|
'KRW-AZTEC', 'KRW-ATH', 'KRW-HOLO', 'KRW-BREV', 'KRW-SHIB',
|
|
]
|
|
|
|
COLLECT_INTERVAL = 3600 # 1시간
|
|
SEARXNG_URL = 'https://searxng.cloud-handson.com/search'
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s %(levelname)s %(message)s',
|
|
handlers=[
|
|
logging.FileHandler('/tmp/context_collector.log'),
|
|
logging.StreamHandler(sys.stdout),
|
|
]
|
|
)
|
|
log = logging.getLogger(__name__)
|
|
|
|
# 코인명 매핑 (검색 키워드용)
|
|
COIN_NAMES = {
|
|
'KRW-BTC': 'Bitcoin', 'KRW-ETH': 'Ethereum', 'KRW-XRP': 'Ripple XRP',
|
|
'KRW-SOL': 'Solana', 'KRW-DOGE': 'Dogecoin', 'KRW-ADA': 'Cardano',
|
|
'KRW-SUI': 'SUI', 'KRW-NEAR': 'NEAR Protocol', 'KRW-KAVA': 'KAVA',
|
|
'KRW-SXP': 'Solar SXP', 'KRW-AKT': 'Akash', 'KRW-SONIC': 'Sonic',
|
|
'KRW-IP': 'Story IP', 'KRW-ORBS': 'ORBS', 'KRW-VIRTUAL': 'Virtuals Protocol',
|
|
'KRW-BARD': 'BARD', 'KRW-XPL': 'XPL', 'KRW-KITE': 'KITE',
|
|
'KRW-ENSO': 'ENSO', 'KRW-0G': '0G', 'KRW-MANTRA': 'MANTRA OM',
|
|
'KRW-EDGE': 'EDGE', 'KRW-CFG': 'Centrifuge', 'KRW-ARDR': 'Ardor',
|
|
'KRW-SIGN': 'SIGN', 'KRW-AZTEC': 'AZTEC', 'KRW-ATH': 'Aethir',
|
|
'KRW-HOLO': 'Holochain', 'KRW-BREV': 'BREV', 'KRW-SHIB': 'Shiba Inu',
|
|
}
|
|
|
|
|
|
def get_conn():
|
|
kwargs = dict(user=os.environ["ORACLE_USER"], password=os.environ["ORACLE_PASSWORD"],
|
|
dsn=os.environ["ORACLE_DSN"])
|
|
if w := os.environ.get("ORACLE_WALLET"):
|
|
kwargs["config_dir"] = w
|
|
return oracledb.connect(**kwargs)
|
|
|
|
|
|
def upsert_context(conn, ticker: str, context_type: str, content: str):
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
"""MERGE INTO ticker_context tc
|
|
USING (SELECT :t AS ticker, :ct AS context_type FROM dual) s
|
|
ON (tc.ticker = s.ticker AND tc.context_type = s.context_type)
|
|
WHEN MATCHED THEN
|
|
UPDATE SET content = :c, updated_at = SYSTIMESTAMP
|
|
WHEN NOT MATCHED THEN
|
|
INSERT (ticker, context_type, content)
|
|
VALUES (:t, :ct, :c)""",
|
|
{'t': ticker, 'ct': context_type, 'c': content},
|
|
)
|
|
|
|
|
|
def collect_price_stats(conn):
|
|
"""각 종목의 24h/7d 가격 변동률 + 거래량 추이를 수집."""
|
|
log.info("[price_stats] 수집 시작")
|
|
count = 0
|
|
for ticker in TICKERS:
|
|
try:
|
|
# 일봉 7개 (7일치)
|
|
df_day = pyupbit.get_ohlcv(ticker, interval='day', count=8)
|
|
if df_day is None or len(df_day) < 2:
|
|
continue
|
|
|
|
today = df_day.iloc[-1]
|
|
yesterday = df_day.iloc[-2]
|
|
|
|
# 24h 변동률
|
|
chg_24h = (today['close'] - yesterday['close']) / yesterday['close'] * 100
|
|
|
|
# 7d 변동률
|
|
if len(df_day) >= 8:
|
|
week_ago = df_day.iloc[-8]
|
|
chg_7d = (today['close'] - week_ago['close']) / week_ago['close'] * 100
|
|
else:
|
|
chg_7d = None
|
|
|
|
# 거래량 추이 (최근 3일 vs 이전 3일)
|
|
if len(df_day) >= 7:
|
|
recent_vol = df_day['volume'].iloc[-3:].mean()
|
|
prev_vol = df_day['volume'].iloc[-6:-3].mean()
|
|
vol_trend = recent_vol / prev_vol if prev_vol > 0 else 1.0
|
|
else:
|
|
vol_trend = None
|
|
|
|
# 1시간봉 최근 24개 (24시간)
|
|
df_h1 = pyupbit.get_ohlcv(ticker, interval='minute60', count=24)
|
|
h1_high = h1_low = None
|
|
if df_h1 is not None and not df_h1.empty:
|
|
h1_high = float(df_h1['high'].max())
|
|
h1_low = float(df_h1['low'].min())
|
|
|
|
stats = {
|
|
'price': float(today['close']),
|
|
'chg_24h_pct': round(chg_24h, 2),
|
|
'chg_7d_pct': round(chg_7d, 2) if chg_7d is not None else None,
|
|
'vol_today': float(today['volume']),
|
|
'vol_trend_3d': round(vol_trend, 2) if vol_trend is not None else None,
|
|
'h24_high': h1_high,
|
|
'h24_low': h1_low,
|
|
'updated': datetime.now().strftime('%Y-%m-%d %H:%M'),
|
|
}
|
|
upsert_context(conn, ticker, 'price_stats', json.dumps(stats, ensure_ascii=False))
|
|
count += 1
|
|
time.sleep(0.2)
|
|
except Exception as e:
|
|
log.warning(f"[price_stats] {ticker} 오류: {e}")
|
|
|
|
conn.commit()
|
|
log.info(f"[price_stats] 완료 {count}/{len(TICKERS)} 종목")
|
|
|
|
|
|
def search_news(coin_name: str, max_results: int = 5) -> list[dict]:
|
|
"""SearXNG로 코인 뉴스 검색."""
|
|
try:
|
|
resp = requests.get(
|
|
SEARXNG_URL,
|
|
params={
|
|
'q': f'{coin_name} crypto news',
|
|
'format': 'json',
|
|
'categories': 'news',
|
|
'language': 'ko-KR',
|
|
'time_range': 'week',
|
|
},
|
|
timeout=15,
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
results = []
|
|
for r in data.get('results', [])[:max_results]:
|
|
results.append({
|
|
'title': r.get('title', ''),
|
|
'url': r.get('url', ''),
|
|
'content': r.get('content', '')[:200],
|
|
'date': r.get('publishedDate', ''),
|
|
})
|
|
return results
|
|
except Exception as e:
|
|
log.warning(f"[news] {coin_name} 검색 오류: {e}")
|
|
return []
|
|
|
|
|
|
def collect_news(conn):
|
|
"""각 종목의 최근 뉴스를 수집."""
|
|
log.info("[news] 수집 시작")
|
|
count = 0
|
|
for ticker in TICKERS:
|
|
coin_name = COIN_NAMES.get(ticker, ticker.split('-')[1])
|
|
try:
|
|
articles = search_news(coin_name)
|
|
if not articles:
|
|
continue
|
|
|
|
news_data = {
|
|
'coin': coin_name,
|
|
'articles': articles,
|
|
'updated': datetime.now().strftime('%Y-%m-%d %H:%M'),
|
|
}
|
|
upsert_context(conn, ticker, 'news', json.dumps(news_data, ensure_ascii=False))
|
|
count += 1
|
|
time.sleep(1) # 검색 API rate limit 배려
|
|
except Exception as e:
|
|
log.warning(f"[news] {ticker} 오류: {e}")
|
|
|
|
conn.commit()
|
|
log.info(f"[news] 완료 {count}/{len(TICKERS)} 종목")
|
|
|
|
|
|
def main():
|
|
log.info("=== context_collector 시작 (1시간 간격) ===")
|
|
log.info(f"대상: {len(TICKERS)}개 종목")
|
|
conn = get_conn()
|
|
|
|
while True:
|
|
t0 = time.time()
|
|
try:
|
|
collect_price_stats(conn)
|
|
collect_news(conn)
|
|
except oracledb.DatabaseError as e:
|
|
log.error(f"DB 오류: {e} — 재연결")
|
|
try:
|
|
conn.close()
|
|
except:
|
|
pass
|
|
conn = get_conn()
|
|
except Exception as e:
|
|
log.error(f"오류: {e}")
|
|
|
|
elapsed = time.time() - t0
|
|
log.info(f"[완료] {elapsed:.0f}초 소요, {COLLECT_INTERVAL}초 후 다음 수집")
|
|
time.sleep(max(60, COLLECT_INTERVAL - elapsed))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|