gahusb/ai-trade
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

LSTM v3 멀티피처, KIS OHLCV 배치, 동적 전략 강화

Browse Source 
- deep_learning.py: INPUT_SIZE=7 (close/open/high/low/volume/rsi/macd),
  feature_scaler/target_scaler 분리, ModelRegistry LRU 종목별 격리 (v3 체크포인트)
- kis.py: get_daily_ohlcv() OHLCV 전체 반환, KISAsyncClient 비동기 배치 조회 추가,
  order() 지정가/조건부 주문 지원
- strategy/process.py: ATR/ADX 기반 동적 손절익절, 트레일링 스탑, 포지션 사이징 강화
- config.py: OLLAMA_NUM_THREAD=8 (9800X3D 최적화), LSTM_COOLDOWN/FAST_EPOCHS 환경변수화
- macro.py: 거시경제 지표 계산 개선
- ollama.py: VRAM 여유량 기반 선택적 언로드
- monitor.py: CPU 서킷 브레이커 연속 횟수 조건 추가
- ipc.py: IPC_STALENESS 600초로 확대
- news.py: 비동기 뉴스 수집 개선
- telegram.py, runner.py: 안정성 개선

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
gahusb
2026-02-24 23:08:33 +09:00
parent 37f6d87bec
commit 4e77a1acf1
11 changed files with 939 additions and 268 deletions
Download Patch File Download Diff File Expand all files Collapse all files

354
modules/services/kis.py
View File

@@ -271,38 +271,48 @@ class KISClient:
except Exception as e:
return {"error": str(e)}
def order(self, ticker, qty, buy_sell, price=0):
"""주문 (시장가)
def order(self, ticker, qty, buy_sell, price=0, order_type="market"):
"""주문
buy_sell: 'BUY' or 'SELL'
order_type: 'market'(시장가), 'limit'(지정가), 'conditional'(조건부지정가)
price: 지정가일 때 주문 가격 (market이면 무시)
"""
self._throttle()
self.ensure_token()
# 모의투자/실전 TR ID 구분
# 매수: VTTC0802U / TTTC0802U
# 매도: VTTC0801U / TTTC0801U
if buy_sell == 'BUY':
tr_id = "VTTC0802U" if self.is_virtual else "TTTC0802U"
else:
tr_id = "VTTC0801U" if self.is_virtual else "TTTC0801U"
# 주문 구분 코드
# 00: 지정가, 01: 시장가, 03: 최유리지정가, 05: 장전시간외, 06: 장후시간외
if order_type == "limit" and price > 0:
ord_dvsn = "00"
ord_unpr = str(int(price))
order_type_str = f"지정가({price:,.0f})"
elif order_type == "conditional" and price > 0:
ord_dvsn = "03" # 최유리지정가
ord_unpr = str(int(price))
order_type_str = f"조건부({price:,.0f})"
else:
ord_dvsn = "01" # 시장가
ord_unpr = "0"
order_type_str = "시장가"
url = f"{self.base_url}/uapi/domestic-stock/v1/trading/order-cash"
# 주문 파라미터
datas = {
"CANO": self.cano,
"ACNT_PRDT_CD": self.acnt_prdt_cd,
"PDNO": ticker,
"ORD_DVSN": "01", # 01: 시장가
"ORD_DVSN": ord_dvsn,
"ORD_QTY": str(qty),
"ORD_UNPR": "0" # 시장가는 0
"ORD_UNPR": ord_unpr
}
# 헤더 준비
headers = self._get_headers(tr_id=tr_id)
# [중요] POST 요청(주문 등) 시 Hash Key 필수
# 단, 모의투자의 경우 일부 상황에서 생략 가능할 수 있으나, 정석대로 포함
hash_key = self.get_hash_key(datas)
if hash_key:
headers["hashkey"] = hash_key
@@ -310,17 +320,17 @@ class KISClient:
print("⚠️ [KIS] Hash Key 생성 실패 (주문 전송 시도)")
try:
print(f"📤 [KIS] 주문 전송: {buy_sell} {ticker} {qty}ea (시장가)")
print(f"📤 [KIS] 주문 전송: {buy_sell} {ticker} {qty}ea ({order_type_str})")
res = requests.post(url, headers=headers, json=datas)
res.raise_for_status()
data = res.json()
print(f"📥 [KIS] 주문 응답 코드(rt_cd): {data['rt_cd']}")
print(f"📥 [KIS] 주문 응답 메시지(msg1): {data['msg1']}")
if data['rt_cd'] != '0':
return {"status": False, "msg": data['msg1'], "rt_cd": data['rt_cd']}
return {"status": True, "msg": "주문 전송 완료", "order_no": data['output']['ODNO'], "rt_cd": data['rt_cd']}
except Exception as e:
return {"status": False, "msg": str(e), "rt_cd": "EXCEPTION"}
@@ -348,34 +358,188 @@ class KISClient:
print(f"❌ 현재가 조회 실패: {e}")
return None
def get_daily_price(self, ticker, period="D"):
"""일별 시세 조회 (기술적 분석용)"""
def _get_daily_ohlcv_by_range(self, ticker, period="D", count=100):
"""기간별시세 API (FHKST03010100) - OHLCV 전체 반환
output2에서 stck_oprc, stck_hgpr, stck_lwpr, stck_clpr, acml_vol 파싱
"""
self._throttle()
self.ensure_token()
end_date = datetime.now().strftime("%Y%m%d")
start_date = (datetime.now() - timedelta(days=int(count * 1.6))).strftime("%Y%m%d")
url = f"{self.base_url}/uapi/domestic-stock/v1/quotations/inquire-daily-itemchartprice"
headers = self._get_headers(tr_id="FHKST03010100")
params = {
"FID_COND_MRKT_DIV_CODE": "J",
"FID_INPUT_ISCD": ticker,
"FID_INPUT_DATE_1": start_date,
"FID_INPUT_DATE_2": end_date,
"FID_PERIOD_DIV_CODE": period,
"FID_ORG_ADJ_PRC": "1"
}
try:
res = requests.get(url, headers=headers, params=params,
timeout=Config.HTTP_TIMEOUT)
res.raise_for_status()
data = res.json()
if data.get('rt_cd') != '0':
return None
output = data.get('output2', [])
if not output:
return None
opens, highs, lows, closes, volumes = [], [], [], [], []
for item in output:
try:
c = int(item.get('stck_clpr', 0) or 0)
o = int(item.get('stck_oprc', 0) or 0)
h = int(item.get('stck_hgpr', 0) or 0)
l = int(item.get('stck_lwpr', 0) or 0)
v = int(item.get('acml_vol', 0) or 0)
if c > 0:
opens.append(o if o > 0 else c)
highs.append(h if h > 0 else c)
lows.append(l if l > 0 else c)
closes.append(c)
volumes.append(v)
except (ValueError, TypeError):
pass
if not closes:
return None
# API는 최신순 → 과거→현재 순으로 변환
opens.reverse(); highs.reverse(); lows.reverse()
closes.reverse(); volumes.reverse()
result = {
'open': opens[-count:],
'high': highs[-count:],
'low': lows[-count:],
'close': closes[-count:],
'volume': volumes[-count:]
}
print(f"[KIS] {ticker} OHLCV: {len(result['close'])}개 ({start_date}~{end_date})")
return result
except Exception as e:
print(f"⚠️ [KIS] OHLCV 조회 실패 ({ticker}): {e}")
return None
def get_daily_ohlcv(self, ticker, period="D", count=100):
"""일별 OHLCV 시세 조회 (기술적 분석 + LSTM 7차원 입력용)
1차: 기간별시세 API OHLCV 파싱 (100일)
2차: 기존 close-only fallback
"""
ohlcv = self._get_daily_ohlcv_by_range(ticker, period, count)
if ohlcv and len(ohlcv['close']) >= 30:
return ohlcv
# fallback: close만 반환 (가짜 OHLCV)
print(f"[KIS] {ticker} OHLCV 실패 → close-only fallback")
prices = self._get_daily_price_by_range(ticker, period, count)
if not prices:
return None
return {
'open': prices, 'high': prices, 'low': prices,
'close': prices, 'volume': []
}
def _get_daily_price_by_range(self, ticker, period="D", count=100):
"""기간별시세 API (FHKST03010100) - 날짜 범위로 최대 100일 데이터 반환
inquire-daily-price(FHKST01010400)가 30일만 반환하는 한계 극복"""
self._throttle()
self.ensure_token()
end_date = datetime.now().strftime("%Y%m%d")
# 영업일 count개 확보를 위해 역일 1.6배 요청 (주말/공휴일 여유)
start_date = (datetime.now() - timedelta(days=int(count * 1.6))).strftime("%Y%m%d")
url = f"{self.base_url}/uapi/domestic-stock/v1/quotations/inquire-daily-itemchartprice"
headers = self._get_headers(tr_id="FHKST03010100")
params = {
"FID_COND_MRKT_DIV_CODE": "J",
"FID_INPUT_ISCD": ticker,
"FID_INPUT_DATE_1": start_date,
"FID_INPUT_DATE_2": end_date,
"FID_PERIOD_DIV_CODE": period,
"FID_ORG_ADJ_PRC": "1"
}
try:
res = requests.get(url, headers=headers, params=params,
timeout=Config.HTTP_TIMEOUT)
res.raise_for_status()
data = res.json()
if data.get('rt_cd') != '0':
return []
# 기간별시세는 output2에 배열로 반환
output = data.get('output2', [])
if not output:
return []
prices = []
for item in output:
clpr = item.get('stck_clpr', '')
if clpr and clpr != '0':
try:
prices.append(int(clpr))
except ValueError:
pass
prices.reverse() # API는 최신순 → 과거→현재 순으로 변환
result = prices[-count:]
print(f"[KIS] {ticker} 기간별시세: {len(result)}개 "
f"({start_date}~{end_date})")
return result
except Exception as e:
print(f"⚠️ [KIS] 기간별시세 조회 실패 ({ticker}): {e}")
return []
def get_daily_price(self, ticker, period="D", count=100):
"""일별 시세 조회 (기술적 분석 + LSTM용)
1차: 기간별시세 API (100일, LSTM 학습 가능)
2차: 구형 API fallback (30일)
"""
# 1차: 기간별시세 API (FHKST03010100) - 100일
prices = self._get_daily_price_by_range(ticker, period, count)
if prices and len(prices) >= 30:
return prices
# 2차: 구형 API fallback (FHKST01010400) - 30일
print(f"[KIS] {ticker} 기간별시세 실패 → 구형 API(30일) fallback")
self._throttle()
self.ensure_token()
url = f"{self.base_url}/uapi/domestic-stock/v1/quotations/inquire-daily-price"
headers = self._get_headers(tr_id="FHKST01010400")
params = {
"FID_COND_MRKT_DIV_CODE": "J",
"FID_INPUT_ISCD": ticker,
"FID_PERIOD_DIV_CODE": period,
"FID_ORG_ADJ_PRC": "1" # 수정주가
"FID_ORG_ADJ_PRC": "1"
}
try:
res = requests.get(url, headers=headers, params=params)
res = requests.get(url, headers=headers, params=params,
timeout=Config.HTTP_TIMEOUT)
res.raise_for_status()
data = res.json()
if data['rt_cd'] != '0':
if data.get('rt_cd') != '0':
return []
# 과거 데이터부터 오도록 정렬 필요할 수 있음 (API는 최신순)
# output 리스트: [ {stck_clpr: 종가, ...}, ... ]
prices = [int(item['stck_clpr']) for item in data['output']]
prices.reverse() # 과거 -> 현재 순으로 정렬
prices = [int(item['stck_clpr']) for item in data['output']
if item.get('stck_clpr')]
prices.reverse()
return prices
except Exception as e:
print(f"❌ 일별 시세 조회 실패: {e}")
print(f"❌ 일별 시세 조회 실패 ({ticker}): {e}")
return []
def get_volume_rank(self, limit=5):
@@ -437,9 +601,18 @@ class KISClient:
data = self._request_api("GET", endpoint, "FHKUP03500100", params=params)
if data['rt_cd'] != '0':
return None
o = data['output']
def _f(val): return float(val) if val else 0.0
def _i(val): return int(float(val)) if val else 0
return {
"price": float(data['output']['bstp_nmix_prpr']), # 현재지수
"change": float(data['output']['bstp_nmix_prdy_ctrt']) # 등락률(%)
"price": _f(o.get('bstp_nmix_prpr')), # 현재지수
"change": _f(o.get('bstp_nmix_prdy_ctrt')), # 등락률(%)
"change_val": _f(o.get('bstp_nmix_prdy_vrss')), # 전일 대비 포인트
"high": _f(o.get('bstp_nmix_hgpr')), # 장중 고가
"low": _f(o.get('bstp_nmix_lwpr')), # 장중 저가
"prev_close": _f(o.get('prdy_nmix')), # 전일 종가
"volume": _i(o.get('acml_vol')), # 누적 거래량(천주)
"trade_value": _i(o.get('acml_tr_pbmn')), # 누적 거래대금(백만원)
}
except Exception as e:
print(f"❌ 지수 조회 실패({ticker}): {e}")
@@ -533,9 +706,8 @@ class KISAsyncClient:
return None
async def get_daily_price_async(self, ticker):
"""비동기 일별 시세 조회"""
"""비동기 일별 시세 조회 (close only, 하위 호환)"""
import aiohttp
import asyncio
self.sync.ensure_token()
url = f"{self.sync.base_url}/uapi/domestic-stock/v1/quotations/inquire-daily-price"
@@ -555,6 +727,52 @@ class KISAsyncClient:
return prices
return []
async def get_daily_ohlcv_async(self, ticker, count=100):
"""비동기 OHLCV 조회 (기간별시세 API 사용)"""
import aiohttp
from datetime import datetime, timedelta
self.sync.ensure_token()
end_date = datetime.now().strftime("%Y%m%d")
start_date = (datetime.now() - timedelta(days=int(count * 1.6))).strftime("%Y%m%d")
url = f"{self.sync.base_url}/uapi/domestic-stock/v1/quotations/inquire-daily-itemchartprice"
headers = self.sync._get_headers(tr_id="FHKST03010100")
params = {
"FID_COND_MRKT_DIV_CODE": "J",
"FID_INPUT_ISCD": ticker,
"FID_INPUT_DATE_1": start_date,
"FID_INPUT_DATE_2": end_date,
"FID_PERIOD_DIV_CODE": "D",
"FID_ORG_ADJ_PRC": "1"
}
async with aiohttp.ClientSession() as session:
data = await self._async_get(session, url, headers, params)
if data and data.get('rt_cd') == '0':
output = data.get('output2', [])
opens, highs, lows, closes, volumes = [], [], [], [], []
for item in output:
try:
c = int(item.get('stck_clpr', 0) or 0)
if c > 0:
opens.append(int(item.get('stck_oprc', 0) or c))
highs.append(int(item.get('stck_hgpr', 0) or c))
lows.append(int(item.get('stck_lwpr', 0) or c))
closes.append(c)
volumes.append(int(item.get('acml_vol', 0) or 0))
except (ValueError, TypeError):
pass
if closes:
opens.reverse(); highs.reverse(); lows.reverse()
closes.reverse(); volumes.reverse()
return {
'open': opens[-count:], 'high': highs[-count:],
'low': lows[-count:], 'close': closes[-count:],
'volume': volumes[-count:]
}
return None
async def get_investor_trend_async(self, ticker):
"""비동기 투자자 동향 조회"""
import aiohttp
@@ -582,7 +800,7 @@ class KISAsyncClient:
return None
async def get_daily_prices_batch(self, tickers):
"""여러 종목의 일별 시세를 병렬로 조회"""
"""여러 종목의 일별 시세(close only)를 병렬로 조회 (하위 호환)"""
import aiohttp
import asyncio
@@ -592,7 +810,6 @@ class KISAsyncClient:
async with aiohttp.ClientSession() as session:
tasks = []
for i, ticker in enumerate(tickers):
# rate limit: 0.5초 간격으로 요청 생성
if i > 0:
await asyncio.sleep(self.min_interval)
@@ -617,6 +834,67 @@ class KISAsyncClient:
return results
async def get_daily_ohlcv_batch(self, tickers, count=100):
"""여러 종목의 OHLCV를 병렬로 조회 (기간별시세 API)"""
import aiohttp
import asyncio
from datetime import datetime, timedelta
self.sync.ensure_token()
results = {}
end_date = datetime.now().strftime("%Y%m%d")
start_date = (datetime.now() - timedelta(days=int(count * 1.6))).strftime("%Y%m%d")
async with aiohttp.ClientSession() as session:
tasks = []
for i, ticker in enumerate(tickers):
if i > 0:
await asyncio.sleep(self.min_interval)
url = f"{self.sync.base_url}/uapi/domestic-stock/v1/quotations/inquire-daily-itemchartprice"
headers = self.sync._get_headers(tr_id="FHKST03010100")
params = {
"FID_COND_MRKT_DIV_CODE": "J",
"FID_INPUT_ISCD": ticker,
"FID_INPUT_DATE_1": start_date,
"FID_INPUT_DATE_2": end_date,
"FID_PERIOD_DIV_CODE": "D",
"FID_ORG_ADJ_PRC": "1"
}
tasks.append((ticker, self._async_get(session, url, headers, params)))
for ticker, task in tasks:
data = await task
if data and data.get('rt_cd') == '0':
output = data.get('output2', [])
opens, highs, lows, closes, volumes = [], [], [], [], []
for item in output:
try:
c = int(item.get('stck_clpr', 0) or 0)
if c > 0:
opens.append(int(item.get('stck_oprc', 0) or c))
highs.append(int(item.get('stck_hgpr', 0) or c))
lows.append(int(item.get('stck_lwpr', 0) or c))
closes.append(c)
volumes.append(int(item.get('acml_vol', 0) or 0))
except (ValueError, TypeError):
pass
if closes:
opens.reverse(); highs.reverse(); lows.reverse()
closes.reverse(); volumes.reverse()
results[ticker] = {
'open': opens[-count:], 'high': highs[-count:],
'low': lows[-count:], 'close': closes[-count:],
'volume': volumes[-count:]
}
else:
results[ticker] = None
else:
results[ticker] = None
return results
async def get_investor_trends_batch(self, tickers):
"""여러 종목의 투자자 동향을 병렬로 조회"""
import aiohttp

37
modules/services/news.py
View File

@@ -28,6 +28,7 @@ class AsyncNewsCollector:
self._cache = None
self._cache_time = 0
self._cache_ttl = 300 # 5분
self._stock_cache = {} # {stock_name: (items, timestamp)}
def get_market_news(self, query="주식 시장"):
"""동기 인터페이스 (하위 호환)"""
@@ -62,11 +63,43 @@ class AsyncNewsCollector:
self._cache_time = now
return items
except ImportError:
# aiohttp 미설치 시 동기 fallback
return self.get_market_news(query)
except Exception as e:
print(f"[News Async] Collection failed: {e}")
# 캐시가 있으면 반환, 없으면 동기 fallback
if self._cache:
return self._cache
return self.get_market_news(query)
async def get_stock_news_async(self, stock_name, max_items=3):
"""종목별 뉴스 수집 (5분 캐싱)
stock_name: 종목 이름 (e.g. '삼성전자', 'SK하이닉스')
"""
now = time.time()
cached = self._stock_cache.get(stock_name)
if cached and (now - cached[1]) < self._cache_ttl:
return cached[0]
try:
import aiohttp
import urllib.parse
query = urllib.parse.quote(f"{stock_name} 주가")
url = f"https://news.google.com/rss/search?q={query}&hl=ko&gl=KR&ceid=KR:ko"
async with aiohttp.ClientSession() as session:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
content = await resp.read()
root = ET.fromstring(content)
items = []
for item in root.findall(".//item")[:max_items]:
title_el = item.find("title")
if title_el is not None and title_el.text:
items.append({"title": title_el.text, "source": "Google News"})
self._stock_cache[stock_name] = (items, now)
return items
except Exception as e:
print(f"[News] 종목 뉴스 수집 실패 ({stock_name}): {e}")
return []
def clear_stock_cache(self):
"""종목 뉴스 캐시 전체 초기화"""
self._stock_cache.clear()

20
modules/services/ollama.py
View File

@@ -113,20 +113,24 @@ class OllamaManager:
"model": self.model_name,
"prompt": prompt,
"stream": False,
"format": "json", # JSON 강제
"format": "json",
"options": {
"num_ctx": 8192, # [5070Ti 최적화] 컨텍스트 크기 2배 증가 (4096 -> 8192)
"temperature": 0.2, # 분석 일관성 유지
"num_gpu": 1, # GPU 사용 명시
"num_thread": 8 # CPU 스레드 수 (9800X3D 활용)
"num_ctx": Config.OLLAMA_NUM_CTX, # 4096 (속도 2배)
"num_predict": Config.OLLAMA_NUM_PREDICT, # 응답 토큰 제한
"temperature": 0.1, # 더 결정론적 (JSON 파싱 안정성)
"num_gpu": 1,
"num_thread": Config.OLLAMA_NUM_THREAD # Config 설정값 (기본 8)
},
"keep_alive": "10m" # [5070Ti 최적화] 10분간 유지 (메모리 여유 있음)
"keep_alive": "5m" # 5분 유지 (불필요한 VRAM 점유 줄임)
}
try:
response = requests.post(self.generate_url, json=payload, timeout=180) # 타임아웃 증가
response = requests.post(self.generate_url, json=payload, timeout=90) # 180→90초
response.raise_for_status()
return response.json().get('response')
except requests.exceptions.Timeout:
print(f"❌ Inference Timeout (90s): {self.model_name}")
return None
except Exception as e:
print(f"❌ Inference Error: {e}")
return None

2
modules/services/telegram.py
View File

@@ -23,7 +23,7 @@ class TelegramMessenger:
payload = {
"chat_id": self.chat_id,
"text": message,
"parse_mode": "Markdown"
"parse_mode": "HTML"
}
try:
requests.post(url, json=payload, timeout=5)

18
modules/services/telegram_bot/runner.py
View File

@@ -30,6 +30,9 @@ def run_telegram_bot_standalone(ipc_lock=None, command_queue=None, shutdown_even
# IPC 초기화 (shared memory + command queue)
ipc = SharedIPC(lock=ipc_lock, command_queue=command_queue)
conflict_retries = 0
MAX_CONFLICT_RETRIES = 10
while True:
# shutdown 체크
if shutdown_event and shutdown_event.is_set():
@@ -51,6 +54,7 @@ def run_telegram_bot_standalone(ipc_lock=None, command_queue=None, shutdown_even
if bot_server.should_restart:
print("[Telegram Bot] Restarting instance...")
conflict_retries = 0 # 정상 재시작 시 카운터 리셋
time.sleep(1)
continue
else:
@@ -61,11 +65,21 @@ def run_telegram_bot_standalone(ipc_lock=None, command_queue=None, shutdown_even
print("[Telegram Bot] Stopped by user")
break
except Exception as e:
if "Conflict" not in str(e):
if "Conflict" in str(e):
conflict_retries += 1
if conflict_retries >= MAX_CONFLICT_RETRIES:
print(f"[Telegram Bot] Conflict max retries ({MAX_CONFLICT_RETRIES}) reached. Exiting.")
break
wait_secs = min(5 * conflict_retries, 30)
print(f"[Telegram Bot] Conflict detected. Waiting {wait_secs}s before retry "
f"({conflict_retries}/{MAX_CONFLICT_RETRIES})...")
time.sleep(wait_secs)
continue
else:
print(f"[Telegram Bot] Error: {e}")
import traceback
traceback.print_exc()
break
break
# 정리
ipc.cleanup()