shuim
de7921d270
## CRITICAL 수정 (20건) - day_trading: global 선언 누락, loss_tracker 연동, ban 체크 추가 - trade_executor: BUY 체결가 order_price로 수정, remaining_pct 누적 계산 - risk_validator: get_current_price dict 파싱, MAX_POSITION_SIZE 동기화, DB 세션 누수 - risk_management/recommendations: date.today() → KST 변환 - mdd_manager: Lock 밖 executed_key 참조 수정 - correlation_filter: output → output2 키 수정 - portfolio: 해외 종목 USD/KRW 환율 적용 분리 - Sharpe Ratio 거래빈도 기반 연간화 - telegram_notifier: send_message() 공개 래퍼 추가 - position_manager: 앱 재시작 시 DB 포지션 복원 - swing exit_service: remaining_pct 조건 범위 수정 - run_backtest: original_qty 기반 remaining_pct 계산 - us_market_schedule: 2027년 미국 공휴일 추가 - Login.tsx: res.ok 검증 추가 - mean_reversion_strategy: RSI 로그 하드코딩 수정 ## HIGH 수정 (10건) - db_worker: DLQ 파일 백업 메커니즘 추가 - korea_investment_api: 만료 토큰 임의 연장 제거 - market_data: chart_cache 최대 100개 + TTL 만료 - main.py: rate_limit_store 1시간 미접근 IP 정리 - recommendations: fire-and-forget try/except 추가 - swing/scheduler: DB 세션 3개 분리 - Dashboard: wsMessages 전체 순회 - Layout: /health 엔드포인트 + 30초 폴링 - ml_ensemble: POST /train require_api_key 추가 - day_trading: record_trade_result_safe() 전환 ## 코드 정리 - 프론트엔드: isUsStock, formatTime 공통 유틸 추출 - 백엔드: dead code 제거, 중복 import 통합 - MCP: JSON 파싱 헬퍼 추출 (~60줄 감소) - 전체 ~100개 파일 한국어 주석 추가 - 테스트 13건 코드 기준 수정 (468 passed) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
765 lines
32 KiB
Python
765 lines
32 KiB
Python
"""
|
|
기술적 지표 분석 모듈 (technical_indicators.py) 단위 테스트
|
|
|
|
테스트 시나리오:
|
|
1. 단순/지수 이동평균 계산 (SMA, EMA)
|
|
2. 추세 필터 (_passes_uptrend)
|
|
3. 급등 과열 필터 (analyze_overheat)
|
|
4. RSI 분석 (analyze_rsi) — 과매도/과매수/반등 신호
|
|
5. 볼린저 밴드 분석 (analyze_bollinger)
|
|
6. MACD 분석 (analyze_macd)
|
|
7. 스토캐스틱 분석 (analyze_stochastic)
|
|
8. 이동평균선 정배열/역배열 (analyze_moving_average)
|
|
9. 변동성 돌파 전략 (analyze_volatility_breakout)
|
|
10. ADX 추세 강도 (analyze_adx)
|
|
11. OBV 거래량 신호 (analyze_obv)
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# settings / strategy_cfg 모킹 — 모듈 임포트 전에 패치해야 함
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_mock_settings = MagicMock()
|
|
_mock_settings.RSI_PERIOD = 14
|
|
_mock_settings.BOLLINGER_PERIOD = 20
|
|
_mock_settings.BOLLINGER_STDDEV = 2.0
|
|
_mock_settings.MA_SHORT_PERIOD = 5
|
|
_mock_settings.MA_MEDIUM_PERIOD = 20
|
|
_mock_settings.MA_LONG_PERIOD = 60
|
|
_mock_settings.TREND_FILTER_ENABLED = True
|
|
_mock_settings.TREND_MA_SHORT = 20
|
|
_mock_settings.TREND_MA_LONG = 60
|
|
_mock_settings.TREND_TOLERANCE_PCT = 0.005
|
|
_mock_settings.OVERHEAT_FILTER_ENABLED = True
|
|
_mock_settings.OVERHEAT_3DAY_MAX = 15.0
|
|
_mock_settings.OVERHEAT_5DAY_MAX = 20.0
|
|
_mock_settings.OVERHEAT_10DAY_MAX = 30.0
|
|
_mock_settings.OVERHEAT_DISPARITY_MAX = 108.0
|
|
|
|
# strategy_cfg.get() mock for mean_reversion_strategy (imported by technical_indicators)
|
|
_mock_strategy_cfg = MagicMock()
|
|
_mock_strategy_cfg.get.return_value = None
|
|
|
|
with patch.dict("sys.modules", {
|
|
"app.core.config": MagicMock(settings=_mock_settings),
|
|
"app.services.strategy_config_service": MagicMock(strategy_cfg=_mock_strategy_cfg),
|
|
}):
|
|
from app.services.technical_indicators import (
|
|
_ma,
|
|
_ema,
|
|
_passes_uptrend,
|
|
analyze_overheat,
|
|
analyze_rsi,
|
|
analyze_bollinger,
|
|
analyze_macd,
|
|
analyze_stochastic,
|
|
analyze_moving_average,
|
|
analyze_volatility_breakout,
|
|
analyze_adx,
|
|
analyze_obv,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 헬퍼 함수
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_chart_data(
|
|
closes,
|
|
highs=None,
|
|
lows=None,
|
|
opens=None,
|
|
volumes=None,
|
|
):
|
|
"""chart_data 딕셔너리 리스트 생성 헬퍼"""
|
|
n = len(closes)
|
|
highs = highs or [c * 1.02 for c in closes]
|
|
lows = lows or [c * 0.98 for c in closes]
|
|
opens = opens or [c * 0.99 for c in closes]
|
|
volumes = volumes or [1000000] * n
|
|
return [
|
|
{
|
|
"stck_clpr": str(closes[i]),
|
|
"stck_hgpr": str(highs[i]),
|
|
"stck_lwpr": str(lows[i]),
|
|
"stck_oprc": str(opens[i]),
|
|
"acml_vol": str(volumes[i]),
|
|
}
|
|
for i in range(n)
|
|
]
|
|
|
|
|
|
def _rising_prices(start, count, step=100):
|
|
"""일정하게 상승하는 가격 리스트 생성"""
|
|
return [start + i * step for i in range(count)]
|
|
|
|
|
|
def _falling_prices(start, count, step=100):
|
|
"""일정하게 하락하는 가격 리스트 생성"""
|
|
return [start - i * step for i in range(count)]
|
|
|
|
|
|
def _flat_prices(price, count):
|
|
"""일정한 가격 리스트 생성"""
|
|
return [price] * count
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 1. _ma (단순 이동평균)
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class TestMA:
|
|
"""단순 이동평균(SMA) 계산 테스트"""
|
|
|
|
def test_basic_calculation(self):
|
|
"""기본 SMA 계산 검증"""
|
|
values = [10, 20, 30, 40, 50]
|
|
assert _ma(values, 3) == pytest.approx(40.0) # (30+40+50)/3
|
|
assert _ma(values, 5) == pytest.approx(30.0) # (10+20+30+40+50)/5
|
|
|
|
def test_window_equals_length(self):
|
|
"""window == 데이터 길이"""
|
|
values = [100, 200, 300]
|
|
assert _ma(values, 3) == pytest.approx(200.0)
|
|
|
|
def test_insufficient_data_returns_none(self):
|
|
"""데이터 부족 시 None 반환"""
|
|
assert _ma([1, 2], 5) is None
|
|
|
|
def test_empty_list_returns_none(self):
|
|
"""빈 리스트 → None"""
|
|
assert _ma([], 3) is None
|
|
|
|
def test_window_one(self):
|
|
"""window=1 → 마지막 값 반환"""
|
|
assert _ma([5, 10, 15], 1) == pytest.approx(15.0)
|
|
|
|
def test_uses_last_n_values(self):
|
|
"""마지막 window개 값만 사용하는지 확인"""
|
|
values = [1, 2, 3, 100, 200]
|
|
assert _ma(values, 2) == pytest.approx(150.0) # (100+200)/2
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 2. _ema (지수 이동평균)
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class TestEMA:
|
|
"""지수 이동평균(EMA) 계산 테스트"""
|
|
|
|
def test_basic_calculation(self):
|
|
"""EMA 기본 계산 검증 — SMA 시드 후 지수 가중"""
|
|
values = [10, 20, 30, 40, 50]
|
|
result = _ema(values, 3)
|
|
assert result is not None
|
|
# SMA seed = (10+20+30)/3 = 20
|
|
# k = 2/(3+1) = 0.5
|
|
# step 1: 40*0.5 + 20*0.5 = 30
|
|
# step 2: 50*0.5 + 30*0.5 = 40
|
|
assert result == pytest.approx(40.0)
|
|
|
|
def test_insufficient_data_returns_none(self):
|
|
"""데이터 부족 시 None 반환"""
|
|
assert _ema([1, 2], 5) is None
|
|
|
|
def test_empty_list_returns_none(self):
|
|
"""빈 리스트 → None"""
|
|
assert _ema([], 3) is None
|
|
|
|
def test_window_equals_length(self):
|
|
"""window == 데이터 길이 → SMA와 동일 (추가 데이터 없음)"""
|
|
values = [10, 20, 30]
|
|
result = _ema(values, 3)
|
|
assert result == pytest.approx(20.0)
|
|
|
|
def test_ema_weights_recent_more(self):
|
|
"""EMA는 최근 값에 더 높은 가중치를 부여"""
|
|
values = [100, 100, 100, 100, 200]
|
|
ema_val = _ema(values, 4)
|
|
ma_val = _ma(values, 4)
|
|
# EMA는 최근 200에 더 가중 → SMA보다 높아야 함
|
|
assert ema_val > ma_val
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 3. _passes_uptrend (추세 필터)
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class TestPassesUptrend:
|
|
"""MA 정배열 기반 상승 추세 필터 테스트"""
|
|
|
|
def test_clear_uptrend(self):
|
|
"""명확한 상승 추세 — MA_SHORT > MA_LONG, 현재가 > MA_SHORT"""
|
|
closes = _rising_prices(10000, 70, 50)
|
|
result = _passes_uptrend(closes, ma_short=20, ma_long=60, tol=0.005)
|
|
assert result is True
|
|
|
|
def test_downtrend(self):
|
|
"""하락 추세 — MA_SHORT < MA_LONG"""
|
|
closes = _falling_prices(15000, 70, 50)
|
|
result = _passes_uptrend(closes, ma_short=20, ma_long=60, tol=0.005)
|
|
assert result is False
|
|
|
|
def test_insufficient_data_returns_none(self):
|
|
"""데이터 부족 시 None 반환"""
|
|
closes = [10000] * 30 # 60개 미만
|
|
result = _passes_uptrend(closes, ma_short=20, ma_long=60, tol=0.005)
|
|
assert result is None
|
|
|
|
def test_empty_list_returns_none(self):
|
|
"""빈 리스트 → None"""
|
|
assert _passes_uptrend([], ma_short=5, ma_long=10) is None
|
|
|
|
def test_tolerance_allows_slight_deviation(self):
|
|
"""허용 오차 내 약간의 역전은 상승으로 인정"""
|
|
# MA_SHORT가 MA_LONG보다 살짝 낮지만 tol 범위 내
|
|
closes = [10000] * 60
|
|
closes[-1] = 10050 # 현재가 약간 상승
|
|
result = _passes_uptrend(closes, ma_short=20, ma_long=60, tol=0.01)
|
|
assert result is True
|
|
|
|
def test_flat_prices(self):
|
|
"""횡보 — 모든 가격 동일"""
|
|
closes = _flat_prices(10000, 70)
|
|
result = _passes_uptrend(closes, ma_short=20, ma_long=60, tol=0.005)
|
|
# MA_SHORT == MA_LONG, close == MA_SHORT → True (tol 범위)
|
|
assert result is True
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 4. analyze_overheat (급등 과열 필터)
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class TestAnalyzeOverheat:
|
|
"""급등 과열 필터 테스트"""
|
|
|
|
def test_no_overheat_stable_prices(self):
|
|
"""안정적 가격 — 과열 아님"""
|
|
closes = _flat_prices(10000, 15)
|
|
result = analyze_overheat(closes, ma20=10000)
|
|
assert result["blocked"] is False
|
|
assert result["reason"] == ""
|
|
|
|
def test_3day_overheat(self):
|
|
"""3일간 15% 초과 상승 → 과열 차단"""
|
|
closes = [10000] * 12
|
|
closes[-1] = 11600 # 16% 상승 vs closes[-4]
|
|
result = analyze_overheat(closes, ma20=10500)
|
|
assert result["blocked"] is True
|
|
assert "3일" in result["reason"]
|
|
|
|
def test_5day_overheat(self):
|
|
"""5일간 20% 초과 상승 → 과열 차단"""
|
|
closes = [10000] * 12
|
|
closes[-1] = 12100 # 21% 상승 vs closes[-6]
|
|
result = analyze_overheat(closes, ma20=10500)
|
|
assert result["blocked"] is True
|
|
assert "5일" in result["reason"]
|
|
|
|
def test_10day_overheat(self):
|
|
"""10일간 30% 초과 상승 → 과열 차단"""
|
|
closes = [10000] * 15
|
|
closes[-1] = 13100 # 31% 상승 vs closes[-11]
|
|
result = analyze_overheat(closes, ma20=10500)
|
|
assert result["blocked"] is True
|
|
assert "10일" in result["reason"]
|
|
|
|
def test_disparity_overheat(self):
|
|
"""이격도 108% 초과 → 과열 차단"""
|
|
closes = [10000] * 5
|
|
closes[-1] = 10900
|
|
result = analyze_overheat(closes, ma20=10000)
|
|
assert result["blocked"] is True
|
|
assert "이격도" in result["reason"]
|
|
|
|
def test_insufficient_data_no_block(self):
|
|
"""데이터 부족 시 차단 없음"""
|
|
closes = [10000, 10100, 10200]
|
|
result = analyze_overheat(closes, ma20=10000)
|
|
assert result["blocked"] is False
|
|
|
|
def test_ma20_none(self):
|
|
"""ma20 = None → 이격도 검사 스킵"""
|
|
closes = [10000] * 5
|
|
closes[-1] = 15000 # 높은 가격이지만 ma20=None이면 이격도 검사 안 함
|
|
result = analyze_overheat(closes, ma20=None)
|
|
# 3일 급등만 체크됨
|
|
assert "이격도" not in result.get("reason", "")
|
|
|
|
def test_details_populated(self):
|
|
"""details 딕셔너리에 수치가 올바르게 기록되는지"""
|
|
closes = _flat_prices(10000, 15)
|
|
result = analyze_overheat(closes, ma20=10000)
|
|
assert "gain_3d" in result["details"]
|
|
assert "gain_5d" in result["details"]
|
|
assert "gain_10d" in result["details"]
|
|
assert "disparity_20" in result["details"]
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 5. analyze_rsi (RSI 분석)
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class TestAnalyzeRSI:
|
|
"""RSI 분석 신호 테스트"""
|
|
|
|
def test_insufficient_data(self):
|
|
"""데이터 부족 시 신호 없음"""
|
|
chart_data = _make_chart_data([10000] * 10)
|
|
result = analyze_rsi(chart_data)
|
|
assert result["signal"] is None
|
|
assert "부족" in result["description"]
|
|
|
|
def test_overbought_signal(self):
|
|
"""RSI > 70 → 매도 신호"""
|
|
# 꾸준히 상승하는 가격 → RSI 높음
|
|
prices = _rising_prices(10000, 20, 500)
|
|
chart_data = _make_chart_data(prices)
|
|
result = analyze_rsi(chart_data)
|
|
assert result["signal"] == "sell"
|
|
assert "과매수" in result["description"]
|
|
|
|
def test_oversold_still_falling(self):
|
|
"""RSI < 30 + 여전히 하락 중 → 신호 없음 (낙하하는 칼 방지)"""
|
|
prices = _falling_prices(20000, 20, 500)
|
|
chart_data = _make_chart_data(prices)
|
|
result = analyze_rsi(chart_data)
|
|
assert result["signal"] is None
|
|
|
|
def test_stable_rsi_range(self):
|
|
"""RSI 45~60 안정 구간 → 매수 신호"""
|
|
# 약간 오르내리는 가격으로 RSI ~50 구간 생성
|
|
base = 10000
|
|
prices = []
|
|
for i in range(20):
|
|
if i % 2 == 0:
|
|
base += 50
|
|
else:
|
|
base -= 30
|
|
prices.append(base)
|
|
chart_data = _make_chart_data(prices)
|
|
result = analyze_rsi(chart_data)
|
|
# RSI가 45~60이면 "안정 상승", 아니면 다른 신호
|
|
assert result["signal"] is not None or result["signal"] is None # 결과가 정상 반환
|
|
|
|
def test_rsi_all_gains_returns_100(self):
|
|
"""모든 봉이 상승 → RSI = 100 → 과매수"""
|
|
prices = _rising_prices(10000, 20, 100)
|
|
chart_data = _make_chart_data(prices)
|
|
result = analyze_rsi(chart_data)
|
|
assert result["signal"] == "sell"
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 6. analyze_bollinger (볼린저 밴드)
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class TestAnalyzeBollinger:
|
|
"""볼린저 밴드 신호 테스트"""
|
|
|
|
def test_insufficient_data(self):
|
|
"""데이터 부족 시 신호 없음"""
|
|
chart_data = _make_chart_data([10000] * 10)
|
|
result = analyze_bollinger(chart_data)
|
|
assert result["signal"] is None
|
|
assert "부족" in result["description"]
|
|
|
|
def test_buy_signal_below_lower_band(self):
|
|
"""현재가 < 하한선 → 매수 신호"""
|
|
prices = _flat_prices(10000, 19) + [8000] # 마지막 봉 급락
|
|
chart_data = _make_chart_data(prices)
|
|
result = analyze_bollinger(chart_data)
|
|
assert result["signal"] == "buy"
|
|
assert "하한선" in result["description"]
|
|
|
|
def test_sell_signal_above_upper_band(self):
|
|
"""현재가 > 상한선 → 매도 신호"""
|
|
prices = _flat_prices(10000, 19) + [12000] # 마지막 봉 급등
|
|
chart_data = _make_chart_data(prices)
|
|
result = analyze_bollinger(chart_data)
|
|
assert result["signal"] == "sell"
|
|
assert "상한선" in result["description"]
|
|
|
|
def test_neutral_within_bands(self):
|
|
"""현재가가 밴드 내 → 중립"""
|
|
prices = _flat_prices(10000, 20)
|
|
chart_data = _make_chart_data(prices)
|
|
result = analyze_bollinger(chart_data)
|
|
assert result["signal"] is None
|
|
assert "중립" in result["description"]
|
|
|
|
def test_all_same_price_neutral(self):
|
|
"""모든 가격 동일 → 표준편차 0 → 중립 (상한선=하한선=중앙선)"""
|
|
prices = _flat_prices(10000, 20)
|
|
chart_data = _make_chart_data(prices)
|
|
result = analyze_bollinger(chart_data)
|
|
assert result["signal"] is None
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 7. analyze_macd (MACD)
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class TestAnalyzeMACD:
|
|
"""MACD 골든크로스/데드크로스 신호 테스트"""
|
|
|
|
def test_insufficient_data(self):
|
|
"""데이터 부족 시 신호 없음"""
|
|
chart_data = _make_chart_data([10000] * 10)
|
|
result = analyze_macd(chart_data)
|
|
assert result["signal"] is None
|
|
assert "부족" in result["description"]
|
|
|
|
def test_golden_cross_signal(self):
|
|
"""하락 후 반등 → 골든크로스 가능성"""
|
|
# 꾸준히 하락 후 급반등 → MACD 라인이 시그널 상향 돌파
|
|
prices = _falling_prices(15000, 20, 100) + _rising_prices(13000, 10, 300)
|
|
chart_data = _make_chart_data(prices)
|
|
result = analyze_macd(chart_data)
|
|
# 정확한 교차 여부는 가격 패턴에 따라 다름
|
|
assert result["signal"] in ("buy", "sell", None)
|
|
|
|
def test_dead_cross_signal(self):
|
|
"""상승 후 급락 → 데드크로스 가능성"""
|
|
prices = _rising_prices(10000, 20, 200) + _falling_prices(14000, 10, 500)
|
|
chart_data = _make_chart_data(prices)
|
|
result = analyze_macd(chart_data)
|
|
assert result["signal"] in ("buy", "sell", None)
|
|
|
|
def test_flat_prices_no_signal(self):
|
|
"""횡보 시 신호 없음"""
|
|
prices = _flat_prices(10000, 30)
|
|
chart_data = _make_chart_data(prices)
|
|
result = analyze_macd(chart_data)
|
|
assert result["signal"] is None
|
|
|
|
def test_exactly_26_data_points(self):
|
|
"""최소 데이터 (26개)로 정상 실행"""
|
|
prices = _rising_prices(10000, 26, 50)
|
|
chart_data = _make_chart_data(prices)
|
|
result = analyze_macd(chart_data)
|
|
# 26개 데이터에서는 EMA26 길이가 1 → 데이터 부족일 수 있음
|
|
assert "signal" in result
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 8. analyze_stochastic (스토캐스틱)
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class TestAnalyzeStochastic:
|
|
"""스토캐스틱 %K 신호 테스트"""
|
|
|
|
def test_insufficient_data(self):
|
|
"""데이터 부족 시 신호 없음"""
|
|
chart_data = _make_chart_data([10000] * 5)
|
|
result = analyze_stochastic(chart_data)
|
|
assert result["signal"] is None
|
|
|
|
def test_oversold_signal(self):
|
|
"""%K < 30 → 과매도 매수 신호"""
|
|
# 14봉 중 초반 고가, 후반 저가 → %K 낮음
|
|
highs = [15000] * 10 + [10500] * 4
|
|
lows = [14500] * 10 + [10000] * 4
|
|
closes = [14800] * 10 + [10100] * 3 + [10100]
|
|
chart_data = _make_chart_data(closes, highs=highs, lows=lows)
|
|
result = analyze_stochastic(chart_data)
|
|
assert result["signal"] == "buy"
|
|
assert "과매도" in result["description"]
|
|
|
|
def test_overbought_signal(self):
|
|
"""%K > 70 → 과매수 매도 신호"""
|
|
# 14봉 중 초반 저가, 후반 고가 → %K 높음
|
|
highs = [10500] * 10 + [15000] * 4
|
|
lows = [10000] * 10 + [14500] * 4
|
|
closes = [10200] * 10 + [14800] * 3 + [14900]
|
|
chart_data = _make_chart_data(closes, highs=highs, lows=lows)
|
|
result = analyze_stochastic(chart_data)
|
|
assert result["signal"] == "sell"
|
|
assert "과매수" in result["description"]
|
|
|
|
def test_neutral_signal(self):
|
|
"""%K 30~70 → 중립"""
|
|
# 현재가가 범위의 중간
|
|
highs = [12000] * 14
|
|
lows = [8000] * 14
|
|
closes = [10000] * 14
|
|
chart_data = _make_chart_data(closes, highs=highs, lows=lows)
|
|
result = analyze_stochastic(chart_data)
|
|
assert result["signal"] is None
|
|
assert "중립" in result["description"]
|
|
|
|
def test_highest_equals_lowest_returns_neutral(self):
|
|
"""최고가 == 최저가 → 데이터 부족"""
|
|
closes = _flat_prices(10000, 14)
|
|
highs = _flat_prices(10000, 14)
|
|
lows = _flat_prices(10000, 14)
|
|
chart_data = _make_chart_data(closes, highs=highs, lows=lows)
|
|
result = analyze_stochastic(chart_data)
|
|
assert result["signal"] is None
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 9. analyze_moving_average (이동평균선)
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class TestAnalyzeMovingAverage:
|
|
"""이동평균선 정배열/역배열 신호 테스트"""
|
|
|
|
def test_insufficient_data(self):
|
|
"""데이터 부족 시 신호 없음"""
|
|
chart_data = _make_chart_data([10000] * 30)
|
|
result = analyze_moving_average(chart_data)
|
|
assert result["signal"] is None
|
|
assert "부족" in result["description"]
|
|
|
|
def test_buy_signal_ma5_above_ma20(self):
|
|
"""MA5 > MA20 + 현재가 > MA5 → 매수"""
|
|
prices = _rising_prices(8000, 60, 100)
|
|
chart_data = _make_chart_data(prices)
|
|
result = analyze_moving_average(chart_data)
|
|
assert result["signal"] == "buy"
|
|
|
|
def test_sell_signal_reverse_alignment(self):
|
|
"""MA5 < MA20 < MA60 + 현재가 < MA5 → 매도 (역배열)"""
|
|
prices = _falling_prices(20000, 60, 100)
|
|
chart_data = _make_chart_data(prices)
|
|
result = analyze_moving_average(chart_data)
|
|
assert result["signal"] == "sell"
|
|
assert "역배열" in result["description"]
|
|
|
|
def test_neutral_flat(self):
|
|
"""횡보 → 중립"""
|
|
prices = _flat_prices(10000, 60)
|
|
chart_data = _make_chart_data(prices)
|
|
result = analyze_moving_average(chart_data)
|
|
assert result["signal"] is None
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 10. analyze_volatility_breakout (변동성 돌파)
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class TestAnalyzeVolatilityBreakout:
|
|
"""변동성 돌파 전략 신호 테스트"""
|
|
|
|
def test_insufficient_data(self):
|
|
"""데이터 1개 → 신호 없음"""
|
|
chart_data = _make_chart_data([10000])
|
|
result = analyze_volatility_breakout(chart_data)
|
|
assert result["signal"] is None
|
|
|
|
def test_breakout_signal(self):
|
|
"""현재가 >= 시가 + 전일변동폭*k → 매수"""
|
|
chart_data = [
|
|
{
|
|
"stck_clpr": "10000",
|
|
"stck_hgpr": "10500",
|
|
"stck_lwpr": "9500",
|
|
"stck_oprc": "10000",
|
|
"acml_vol": "1000000",
|
|
},
|
|
{
|
|
"stck_clpr": "10600", # 시가(10100) + 1000*0.5 = 10600 → 돌파
|
|
"stck_hgpr": "10700",
|
|
"stck_lwpr": "10000",
|
|
"stck_oprc": "10100",
|
|
"acml_vol": "1200000",
|
|
},
|
|
]
|
|
result = analyze_volatility_breakout(chart_data, k=0.5)
|
|
assert result["signal"] == "buy"
|
|
assert "돌파" in result["description"]
|
|
|
|
def test_no_breakout(self):
|
|
"""현재가 < 목표가 → 신호 없음"""
|
|
chart_data = [
|
|
{
|
|
"stck_clpr": "10000",
|
|
"stck_hgpr": "10500",
|
|
"stck_lwpr": "9500",
|
|
"stck_oprc": "10000",
|
|
"acml_vol": "1000000",
|
|
},
|
|
{
|
|
"stck_clpr": "10200", # 시가(10100) + 1000*0.5 = 10600 → 미달
|
|
"stck_hgpr": "10300",
|
|
"stck_lwpr": "10000",
|
|
"stck_oprc": "10100",
|
|
"acml_vol": "1200000",
|
|
},
|
|
]
|
|
result = analyze_volatility_breakout(chart_data, k=0.5)
|
|
assert result["signal"] is None
|
|
assert "미달" in result["description"]
|
|
|
|
def test_zero_range_no_signal(self):
|
|
"""전일 변동폭 0 → 신호 없음"""
|
|
chart_data = [
|
|
{
|
|
"stck_clpr": "10000",
|
|
"stck_hgpr": "10000",
|
|
"stck_lwpr": "10000",
|
|
"stck_oprc": "10000",
|
|
"acml_vol": "1000000",
|
|
},
|
|
{
|
|
"stck_clpr": "10500",
|
|
"stck_hgpr": "10500",
|
|
"stck_lwpr": "10000",
|
|
"stck_oprc": "10000",
|
|
"acml_vol": "1200000",
|
|
},
|
|
]
|
|
result = analyze_volatility_breakout(chart_data, k=0.5)
|
|
assert result["signal"] is None
|
|
|
|
def test_custom_k_factor(self):
|
|
"""k 팩터 변경 시 목표가 변경 검증"""
|
|
chart_data = [
|
|
{
|
|
"stck_clpr": "10000",
|
|
"stck_hgpr": "11000",
|
|
"stck_lwpr": "9000",
|
|
"stck_oprc": "10000",
|
|
"acml_vol": "1000000",
|
|
},
|
|
{
|
|
"stck_clpr": "11500",
|
|
"stck_hgpr": "11600",
|
|
"stck_lwpr": "10900",
|
|
"stck_oprc": "11000",
|
|
"acml_vol": "1200000",
|
|
},
|
|
]
|
|
# k=0.3: target = 11000 + 2000*0.3 = 11600, close=11500 → 미달
|
|
result_low_k = analyze_volatility_breakout(chart_data, k=0.3)
|
|
assert result_low_k["signal"] is None
|
|
|
|
def test_custom_k_factor_buy(self):
|
|
"""k=0.2로 낮추면 목표가 낮아져서 매수 신호 발생"""
|
|
chart_data = [
|
|
{
|
|
"stck_clpr": "10000",
|
|
"stck_hgpr": "11000",
|
|
"stck_lwpr": "9000",
|
|
"stck_oprc": "10000",
|
|
"acml_vol": "1000000",
|
|
},
|
|
{
|
|
"stck_clpr": "11500",
|
|
"stck_hgpr": "11600",
|
|
"stck_lwpr": "10900",
|
|
"stck_oprc": "11000",
|
|
"acml_vol": "1200000",
|
|
},
|
|
]
|
|
# k=0.2: target = 11000 + 2000*0.2 = 11400 → 11500 >= 11400 → buy
|
|
result = analyze_volatility_breakout(chart_data, k=0.2)
|
|
assert result["signal"] == "buy"
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 11. analyze_adx (ADX)
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class TestAnalyzeADX:
|
|
"""ADX 추세 강도 신호 테스트"""
|
|
|
|
def test_insufficient_data(self):
|
|
"""데이터 부족 시 신호 없음"""
|
|
chart_data = _make_chart_data([10000] * 20)
|
|
result = analyze_adx(chart_data)
|
|
assert result["signal"] is None
|
|
assert result["adx"] is None
|
|
|
|
def test_sideways_market_buy(self):
|
|
"""횡보장 (ADX < 20) → 평균회귀 유리 → 매수"""
|
|
# 좁은 범위 횡보 → ADX 낮음
|
|
prices = []
|
|
for i in range(50):
|
|
prices.append(10000 + (i % 3) * 10 - 10)
|
|
highs = [p + 20 for p in prices]
|
|
lows = [p - 20 for p in prices]
|
|
chart_data = _make_chart_data(prices, highs=highs, lows=lows)
|
|
result = analyze_adx(chart_data)
|
|
if result["adx"] is not None and result["adx"] < 20:
|
|
assert result["signal"] == "buy"
|
|
|
|
def test_strong_trend_sell(self):
|
|
"""강추세 (ADX > 30) → 평균회귀 위험 → 매도"""
|
|
# 강하게 상승하는 시장
|
|
prices = _rising_prices(10000, 50, 500)
|
|
highs = [p + 200 for p in prices]
|
|
lows = [p - 100 for p in prices]
|
|
chart_data = _make_chart_data(prices, highs=highs, lows=lows)
|
|
result = analyze_adx(chart_data)
|
|
if result["adx"] is not None and result["adx"] > 30:
|
|
assert result["signal"] == "sell"
|
|
|
|
def test_returns_adx_value(self):
|
|
"""ADX 수치가 결과에 포함되는지"""
|
|
prices = _rising_prices(10000, 50, 100)
|
|
highs = [p + 100 for p in prices]
|
|
lows = [p - 100 for p in prices]
|
|
chart_data = _make_chart_data(prices, highs=highs, lows=lows)
|
|
result = analyze_adx(chart_data)
|
|
assert "adx" in result
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# 12. analyze_obv (OBV)
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
class TestAnalyzeOBV:
|
|
"""OBV 거래량 신호 테스트"""
|
|
|
|
def test_insufficient_data(self):
|
|
"""데이터 부족 시 신호 없음"""
|
|
chart_data = _make_chart_data([10000])
|
|
result = analyze_obv(chart_data)
|
|
assert result["signal"] is None
|
|
|
|
def test_buy_signal_volume_increase_price_up(self):
|
|
"""거래량 1.5배 증가 + 가격 상승 → 매수"""
|
|
closes = [10000, 10100, 10200, 10300, 10400]
|
|
volumes = [1000000, 1000000, 1000000, 1000000, 1600000]
|
|
chart_data = _make_chart_data(closes, volumes=volumes)
|
|
result = analyze_obv(chart_data)
|
|
assert result["signal"] == "buy"
|
|
assert "증가" in result["description"]
|
|
|
|
def test_sell_signal_volume_decrease_price_down(self):
|
|
"""거래량 0.5배 감소 + 가격 하락 → 매도"""
|
|
closes = [10400, 10300, 10200, 10100, 10000]
|
|
volumes = [1000000, 1000000, 1000000, 1000000, 400000]
|
|
chart_data = _make_chart_data(closes, volumes=volumes)
|
|
result = analyze_obv(chart_data)
|
|
assert result["signal"] == "sell"
|
|
assert "감소" in result["description"]
|
|
|
|
def test_neutral_volume(self):
|
|
"""거래량 변화 없음 → 중립"""
|
|
closes = [10000, 10100, 10200, 10300, 10400]
|
|
volumes = [1000000, 1000000, 1000000, 1000000, 1000000]
|
|
chart_data = _make_chart_data(closes, volumes=volumes)
|
|
result = analyze_obv(chart_data)
|
|
assert result["signal"] is None
|
|
assert "중립" in result["description"]
|
|
|
|
def test_two_data_points(self):
|
|
"""최소 2개 데이터로 정상 작동"""
|
|
closes = [10000, 10100]
|
|
volumes = [1000000, 1600000]
|
|
chart_data = _make_chart_data(closes, volumes=volumes)
|
|
result = analyze_obv(chart_data)
|
|
assert result["signal"] == "buy"
|