Files
auto-stock/backend/app/services/market_analyzer.py
2026-03-06 20:46:08 +09:00

564 lines
22 KiB
Python

"""
시장 분석 서비스
- KIS API를 통해 관심 종목(watchlist)의 기술적 지표 일괄 분석
- 시장 국면(bull/bear/sideways) 판별 및 추천 종목 선별
- OVERHEAT(급등 과열) 필터, 추세 필터 적용
- 분석 결과를 StockAnalysis DB에 upsert하고 Redis 캐시에 저장
"""
import asyncio
import logging
import time
import pytz
from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel
from app.core.config import settings
from app.core.market_schedule import is_market_open
from app.services.korea_investment_api import korea_investment_api, real_market_api as _real_market_api
from app.services.mean_reversion_strategy import (
MarketRegime, detect_market_regime, calculate_atr, calculate_adx
)
from app.services.technical_indicators import (
TREND_FILTER_ENABLED, TREND_MA_SHORT, TREND_MA_LONG, TREND_TOLERANCE_PCT,
OVERHEAT_FILTER_ENABLED, _passes_uptrend, analyze_overheat,
analyze_custom_signals, analyze_macd, analyze_rsi, analyze_bollinger,
analyze_moving_average, analyze_stochastic, analyze_volatility_breakout,
analyze_adx, analyze_obv
)
logger = logging.getLogger(__name__)
KST = pytz.timezone('Asia/Seoul')
def _get_kst_now():
return datetime.now(KST).replace(tzinfo=None)
class RecommendationCriteria(BaseModel):
use_macd: bool = False
use_rsi: bool = False
use_bollinger: bool = False
use_moving_average: bool = False
use_stochastic: bool = False
use_adx: bool = False
use_obv: bool = False
use_volatility_breakout: bool = False
use_overheat_filter: bool = True
min_volume_ratio: float = 1.0
max_disparity: float = 200.0
for_batch: bool = False
class StockRecommendation(BaseModel):
stock_code: str
stock_name: str
current_price: float
change_rate: float
signals: List[str]
score: int
price_date: str = ""
volume: Optional[int] = None
rsi: Optional[float] = None
ma5: Optional[float] = None
ma20: Optional[float] = None
ma60: Optional[float] = None
stochastic_k: Optional[float] = None
stochastic_d: Optional[float] = None
bollinger_upper: Optional[float] = None
bollinger_lower: Optional[float] = None
bollinger_middle: Optional[float] = None
ml_score: Optional[int] = None
ml_confidence: Optional[float] = None
ml_signal: Optional[str] = None
atr_14: Optional[float] = None
adx_14: Optional[float] = None
_market_regime_cache: dict = {"result": None, "ts": 0.0}
_MARKET_REGIME_CACHE_TTL = 120
async def check_market_regime() -> dict:
try:
if not is_market_open():
return {"regime": "neutral", "regime_200d": "bear", "message": "장외 시간 - 시장 레짐 판별 스킵", "position_multiplier": 0.5}
now_ts = time.monotonic()
if _market_regime_cache["result"] is not None and (now_ts - _market_regime_cache["ts"]) < _MARKET_REGIME_CACHE_TTL:
return _market_regime_cache["result"]
_api = _real_market_api or korea_investment_api
kospi_chart = await _api.get_daily_chart("069500", "D", limit=240)
if "output2" not in kospi_chart or not kospi_chart["output2"]:
return {"regime": "neutral", "message": "시장 지수 확인 불가", "position_multiplier": 0.5}
chart_data = kospi_chart["output2"]
closes = [float(item.get("stck_clpr", 0)) for item in chart_data]
if len(closes) < 60:
return {"regime": "neutral", "message": "데이터 부족", "position_multiplier": 0.5}
def _cache_and_return(result: dict) -> dict:
_market_regime_cache["result"] = result
_market_regime_cache["ts"] = time.monotonic()
return result
# ── 수동 비상 정지 ────────────────────────────────────────────
if settings.TRADING_HALTED:
return _cache_and_return({
"regime": "halted",
"regime_200d": "bear",
"message": "⛔ TRADING_HALTED=true — 수동 비상 정지 중 (매수 전면 차단)",
"position_multiplier": 0.0,
"min_score": 9999,
"max_trades": 0,
})
current_price = closes[0]
ma20 = sum(closes[:20]) / 20
ma60 = sum(closes[:60]) / 60
if len(closes) >= 205:
regime_enum = detect_market_regime(list(reversed(closes)))
is_bull_200 = regime_enum == MarketRegime.BULL
ma200 = sum(closes[:200]) / 200
else:
is_bull_200 = current_price > ma20
ma200 = ma60
# ── 당일 급락 감지 (사이드카/전쟁 등) ───────────────────────
# closes[0]이 오늘 데이터일 때만 적용 (일봉 미확정 시 전일 종가가 closes[0]이 되어 오탐 발생)
latest_date = chart_data[0].get("stck_bsop_date", "") if chart_data else ""
today_str = _get_kst_now().strftime("%Y%m%d")
if latest_date == today_str and len(closes) >= 2 and closes[1] > 0:
today_change_pct = (closes[0] - closes[1]) / closes[1] * 100
if today_change_pct <= settings.MARKET_CRASH_HALT_PCT:
return _cache_and_return({
"regime": "crash",
"regime_200d": "bear",
"message": f"🚨 당일 급락 감지 ({today_change_pct:.1f}% ≤ {settings.MARKET_CRASH_HALT_PCT}%) — 매수 전면 차단",
"position_multiplier": 0.0,
"min_score": 9999,
"max_trades": 0,
"today_change_pct": round(today_change_pct, 2),
})
consecutive_down = 0
for i in range(len(closes) - 1):
if closes[i] < closes[i + 1]:
consecutive_down += 1
else:
break
drop_from_ma20 = ((current_price - ma20) / ma20) * 100
if current_price < ma60 or consecutive_down >= 5:
return _cache_and_return({
"regime": "strong_bearish",
"regime_200d": "bear",
"message": f"강한 하락장 (코스피: {current_price:.0f}, MA200: {ma200:.0f}, {consecutive_down}일 연속하락)",
"position_multiplier": 0.1,
"min_score": settings.BEAR_MARKET_MIN_SCORE + 3,
"max_trades": 1,
"consecutive_down": consecutive_down,
"drop_from_ma20": round(drop_from_ma20, 1),
})
if current_price < ma20:
return _cache_and_return({
"regime": "bearish",
"regime_200d": "bear",
"message": f"하락장 (코스피: {current_price:.0f} < MA20: {ma20:.0f}, MA200: {ma200:.0f})",
"position_multiplier": settings.BEAR_MARKET_POSITION_RATIO,
"min_score": settings.BEAR_MARKET_MIN_SCORE,
"max_trades": settings.BEAR_MARKET_MAX_TRADES,
"consecutive_down": consecutive_down,
"drop_from_ma20": round(drop_from_ma20, 1),
})
return _cache_and_return({
"regime": "bullish",
"regime_200d": "bull" if is_bull_200 else "bear",
"message": f"상승장 (코스피: {current_price:.0f}, MA200: {ma200:.0f}, Bull={is_bull_200})",
"position_multiplier": 1.0,
"min_score": 1,
"max_trades": settings.MAX_DAILY_TRADES,
"consecutive_down": consecutive_down,
"drop_from_ma20": round(drop_from_ma20, 1),
})
except Exception as e:
logger.error(f"시장 지수 확인 실패: {e}")
return {"regime": "neutral", "message": "시장 지수 확인 실패", "position_multiplier": 0.5}
def _analyze_chart_items(
chart_data: List[dict],
stock_code: str,
stock_name: str,
criteria: "RecommendationCriteria",
) -> Optional[StockRecommendation]:
"""KIS 형식 chart_data (oldest first)로 기술적 분석 수행 — API 호출 없음.
analyze_stock / analyze_stock_for_batch_db 양쪽에서 공유하는 순수 분석 로직.
"""
latest = chart_data[-1] if chart_data else None
if not latest:
return None
if not stock_name:
stock_name = latest.get("hts_kor_isnm", stock_code)
current_price = float(latest.get("stck_clpr", 0))
price_date = latest.get("stck_bsop_date", "")
volume = int(latest.get("acml_vol", 0) or 0)
if current_price == 0:
logger.debug("필터[가격0] %s", stock_code)
return None
# 가격 필터: 5,000원 미만 종목 제외 (CLAUDE.md 섹션 5.4)
if current_price < settings.MIN_PRICE_KRW:
logger.info("필터[저가] %s: %.0f원 < %d원", stock_code, current_price, settings.MIN_PRICE_KRW)
return None
# 전일 거래량 필터: 100만주 미만 종목 제외 (유동성 부족)
if volume < 1_000_000:
logger.info("필터[거래량] %s: %d주 < 100만주", stock_code, volume)
return None
if len(chart_data) >= 2:
prev_price = float(chart_data[-2].get("stck_clpr", current_price))
change_rate = ((current_price - prev_price) / prev_price) * 100 if prev_price > 0 else 0
else:
change_rate = 0
closes = [float(item.get("stck_clpr", 0)) for item in chart_data]
volumes = [int(item.get("acml_vol", 0)) for item in chart_data]
ma5 = sum(closes[-5:]) / 5 if len(closes) >= 5 else None
ma20 = sum(closes[-20:]) / 20 if len(closes) >= 20 else None
ma60 = sum(closes[-60:]) / 60 if len(closes) >= 60 else None
avg_vol_20 = sum(volumes[-20:]) / 20 if len(volumes) >= 20 else 0
volume_ratio = (volume / avg_vol_20) if avg_vol_20 > 0 else 0
if not criteria.for_batch and avg_vol_20 < 50000 and current_price < 50000:
return None
disparity = (current_price / ma20) * 100 if ma20 else 100
trend_ok = None
if TREND_FILTER_ENABLED:
trend_ok = _passes_uptrend(closes, TREND_MA_SHORT, TREND_MA_LONG)
rsi = None
if len(closes) >= 15:
prices_for_rsi = closes[-15:]
gains = []
losses = []
for i in range(1, len(prices_for_rsi)):
diff = prices_for_rsi[i] - prices_for_rsi[i-1]
gains.append(max(diff, 0))
losses.append(max(-diff, 0))
avg_gain = sum(gains[-14:]) / 14
avg_loss = sum(losses[-14:]) / 14
if avg_loss == 0:
rsi = 100.0
else:
rs = avg_gain / avg_loss
rsi = 100 - (100 / (1 + rs))
stochastic_k = None
stochastic_d = None
if len(chart_data) >= 14:
recent = chart_data[-14:]
highs = [float(item.get("stck_hgpr", 0)) for item in recent]
lows = [float(item.get("stck_lwpr", 0)) for item in recent]
closes_for_stoch = [float(item.get("stck_clpr", 0)) for item in recent]
highest_high = max(highs)
lowest_low = min(lows)
if highest_high != lowest_low:
stochastic_k = ((closes_for_stoch[-1] - lowest_low) / (highest_high - lowest_low)) * 100
stochastic_d_vals = []
for j in range(len(recent)-3, len(recent)):
slice_high = max([float(item.get("stck_hgpr", 0)) for item in recent[j-3 if j-3>=0 else 0:j+1]])
slice_low = min([float(item.get("stck_lwpr", 0)) for item in recent[j-3 if j-3>=0 else 0:j+1]])
if slice_high != slice_low:
stochastic_d_vals.append(((float(recent[j].get("stck_clpr", 0)) - slice_low) / (slice_high - slice_low)) * 100)
if stochastic_d_vals:
stochastic_d = sum(stochastic_d_vals) / len(stochastic_d_vals)
bollinger_upper = bollinger_lower = bollinger_middle = None
if len(closes) >= 20:
window = closes[-20:]
middle = sum(window) / len(window)
variance = sum((p - middle) ** 2 for p in window) / len(window)
stddev = variance ** 0.5
bollinger_middle = middle
bollinger_upper = middle + 2 * stddev
bollinger_lower = middle - 2 * stddev
signals = []
score = 0
if disparity > criteria.max_disparity:
score -= 5
signals.append(f"이격도 과열({disparity:.1f}% > {criteria.max_disparity}%) - 매수금지")
if criteria.use_overheat_filter and OVERHEAT_FILTER_ENABLED:
overheat = analyze_overheat(closes, ma20)
if overheat["blocked"]:
score -= 10
signals.append(overheat["reason"])
if criteria.use_volatility_breakout:
vb_result = analyze_volatility_breakout(chart_data)
if vb_result and vb_result["signal"] == "buy":
signals.append(vb_result["description"])
score += 4
custom_result = analyze_custom_signals(chart_data)
if custom_result:
if custom_result["signal"] == "buy":
signals.append(custom_result["description"])
score += 3
elif custom_result["signal"] == "sell":
signals.append(custom_result["description"])
score -= 2
elif custom_result["signal"] == "exit":
signals.append(custom_result["description"])
score -= 5
elif custom_result["signal"] == "take_profit":
signals.append(custom_result["description"])
score += 2
if criteria.use_macd:
result = analyze_macd(chart_data)
if result["signal"] == "buy":
signals.append(result["description"])
score += 2
if criteria.use_rsi:
result = analyze_rsi(chart_data)
if result["signal"] == "buy":
if volume_ratio >= 1.2:
signals.append(result["description"] + " + 거래량증가")
score += 3
else:
signals.append(result["description"] + " (거래량부족)")
score += 1
if criteria.use_bollinger:
result = analyze_bollinger(chart_data)
if result["signal"] == "buy":
signals.append(result["description"])
score += 1
if criteria.use_moving_average:
result = analyze_moving_average(chart_data)
if result["signal"] == "buy":
signals.append(result["description"])
score += 2
if criteria.use_stochastic:
result = analyze_stochastic(chart_data)
if result["signal"] == "buy":
signals.append(result["description"])
score += 1
if criteria.use_adx:
result = analyze_adx(chart_data)
if result["signal"] == "buy":
signals.append(result["description"])
score += 1
if criteria.use_obv:
result = analyze_obv(chart_data)
if result["signal"] == "buy":
signals.append(result["description"])
score += 1
if volume_ratio >= 1.5 and change_rate > 0:
signals.append(f"거래량 실린 상승 ({volume_ratio:.1f}배)")
score += 3
elif volume_ratio < 0.8:
score -= 1
if TREND_FILTER_ENABLED and trend_ok is not None:
if trend_ok:
signals.append(f"추세상승(>{TREND_MA_SHORT}MA & {TREND_MA_SHORT}MA>{TREND_MA_LONG}MA, tol={int(TREND_TOLERANCE_PCT*10000)/100:.2f}%)")
score += 1
else:
signals.append(f"추세약함({TREND_MA_SHORT}MA≤{TREND_MA_LONG}MA 또는 종가≤{TREND_MA_SHORT}MA)")
if criteria.min_volume_ratio > 0 and volume_ratio < criteria.min_volume_ratio:
return None
ml_score_val = None
ml_confidence_val = None
ml_signal_val = None
if settings.ML_ENSEMBLE_ENABLED:
try:
from app.services.ml_ensemble import get_ensemble
ensemble = get_ensemble()
if ensemble.is_trained:
ml_result = ensemble.predict(chart_data)
if "error" not in ml_result:
ml_score_val = ml_result["ensemble_score"]
ml_confidence_val = ml_result["confidence"]
ml_signal_val = ml_result["signal"]
rule_score_normalized = (score / 15.0) * 100.0
combined_score = int(
rule_score_normalized * settings.ML_ENSEMBLE_RULE_WEIGHT +
ml_score_val * settings.ML_ENSEMBLE_ML_WEIGHT
)
score = max(0, int(combined_score / 100.0 * 15.0))
signals.append(f"ML앙상블: {ml_signal_val}({ml_score_val}점, 신뢰도{ml_confidence_val:.0%})")
except Exception:
pass
atr_14_val = None
adx_14_val = None
try:
if len(chart_data) >= 15:
_h = [float(d.get("stck_hgpr", 0)) for d in chart_data]
_l = [float(d.get("stck_lwpr", 0)) for d in chart_data]
_c = [float(d.get("stck_clpr", 0)) for d in chart_data]
atr_14_val = calculate_atr(_h, _l, _c)
if len(chart_data) >= 29:
adx_14_val = calculate_adx(_h, _l, _c)
except Exception as _e:
logger.debug(f"ATR/ADX 계산 실패 {stock_code}: {_e}")
if criteria.for_batch or len(signals) > 0 or score > 0:
final_score = max(1, score) if len(signals) > 0 else score
return StockRecommendation(
stock_code=stock_code,
stock_name=stock_name,
current_price=current_price,
change_rate=change_rate,
signals=signals if signals else ["조건 분석 중"],
score=final_score,
price_date=price_date,
volume=volume,
rsi=rsi,
ma5=ma5,
ma20=ma20,
ma60=ma60,
stochastic_k=stochastic_k,
stochastic_d=stochastic_d,
bollinger_upper=bollinger_upper,
bollinger_lower=bollinger_lower,
bollinger_middle=bollinger_middle,
ml_score=ml_score_val,
ml_confidence=ml_confidence_val,
ml_signal=ml_signal_val,
atr_14=atr_14_val,
adx_14=adx_14_val,
)
return None
async def analyze_stock(stock_code: str, stock_name: str, criteria: RecommendationCriteria) -> Optional[StockRecommendation]:
"""KIS API로 차트를 조회한 뒤 _analyze_chart_items에 위임."""
try:
_api = _real_market_api or korea_investment_api
chart_response = await _api.get_daily_chart(stock_code, "D")
if "output2" not in chart_response or not chart_response["output2"]:
return None
chart_data = list(reversed(chart_response["output2"]))
return _analyze_chart_items(chart_data, stock_code, stock_name, criteria)
except Exception as e:
logger.error(f"Error analyzing {stock_code}: {str(e)}")
return None
async def analyze_stock_for_batch_db(
chart_data_newest_first: List[dict],
stock_code: str,
stock_name: str,
) -> Optional[StockRecommendation]:
"""DB에서 가져온 chart_data (newest first)로 배치 분석 — KIS API 호출 없음."""
criteria = RecommendationCriteria(
use_macd=True,
use_rsi=True,
use_bollinger=True,
use_moving_average=True,
use_stochastic=True,
use_adx=True,
use_obv=True,
use_volatility_breakout=True,
min_volume_ratio=0,
max_disparity=200.0,
for_batch=True,
)
try:
chart_data = list(reversed(chart_data_newest_first)) # oldest first
return _analyze_chart_items(chart_data, stock_code, stock_name, criteria)
except Exception as e:
logger.error(f"Error analyzing {stock_code}: {str(e)}")
return None
async def analyze_stock_for_batch(stock_code: str, stock_name: str) -> Optional[StockRecommendation]:
criteria = RecommendationCriteria(
use_macd=True,
use_rsi=True,
use_bollinger=True,
use_moving_average=True,
use_stochastic=True,
use_adx=True,
use_obv=True,
use_volatility_breakout=True,
min_volume_ratio=0,
max_disparity=200.0,
for_batch=True,
)
return await analyze_stock(stock_code, stock_name, criteria)
async def analyze_stock_with_indicators(stock_code: str, stock_name: str) -> Optional[StockRecommendation]:
criteria = RecommendationCriteria(
use_macd=True,
use_rsi=True,
use_bollinger=True,
use_moving_average=True,
use_stochastic=True,
use_adx=True,
use_obv=True,
use_volatility_breakout=True,
min_volume_ratio=1.2,
max_disparity=110.0
)
return await analyze_stock(stock_code, stock_name, criteria)
async def get_all_tradable_stocks() -> List[tuple]:
SKIP_KEYWORDS = [
"ETF", "레버리지", "인버스", "KODEX", "TIGER", "ARIRANG",
"KBSTAR", "HANARO", "SOL", "ACE", "KOSEF", "KINDEX", "RISE",
"선물", "스팩", "리츠", "SPAC",
]
try:
_list_api = _real_market_api or korea_investment_api
await asyncio.sleep(3.0)
kospi_result = await _list_api.get_stock_list("J")
await asyncio.sleep(3.0)
kosdaq_result = await _list_api.get_stock_list("Q")
stocks = []
for result in (kospi_result, kosdaq_result):
if "output" in result:
for item in result["output"]:
stock_code = item.get("mksc_shrn_iscd", "")
stock_name = item.get("hts_kor_isnm", "")
if not stock_code or not stock_name:
continue
if any(kw in stock_name for kw in SKIP_KEYWORDS):
continue
stocks.append((stock_code, stock_name))
return stocks
except Exception as e:
logger.error(f"Error fetching stock list from KIS API: {e}", exc_info=True)
return []