128 lines
4.1 KiB
Python
128 lines
4.1 KiB
Python
"""
|
|
E2E tests for KJB strategy flow.
|
|
"""
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
def test_kjb_strategy_endpoint(client: TestClient, auth_headers):
|
|
"""Test KJB strategy ranking endpoint."""
|
|
response = client.post(
|
|
"/api/strategy/kjb",
|
|
json={
|
|
"universe": {"markets": ["KOSPI"]},
|
|
"top_n": 10,
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code in [200, 400, 500]
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
assert data["strategy_name"] == "kjb"
|
|
assert "stocks" in data
|
|
|
|
|
|
def test_kjb_backtest_creation(client: TestClient, auth_headers):
|
|
"""Test creating a KJB backtest."""
|
|
response = client.post(
|
|
"/api/backtest",
|
|
json={
|
|
"strategy_type": "kjb",
|
|
"strategy_params": {
|
|
"max_positions": 10,
|
|
"cash_reserve_ratio": 0.3,
|
|
"stop_loss_pct": 0.03,
|
|
"target1_pct": 0.05,
|
|
"target2_pct": 0.10,
|
|
},
|
|
"start_date": "2023-01-01",
|
|
"end_date": "2023-12-31",
|
|
"initial_capital": 10000000,
|
|
"commission_rate": 0.00015,
|
|
"slippage_rate": 0.001,
|
|
"benchmark": "KOSPI",
|
|
"top_n": 30,
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "id" in data
|
|
assert data["status"] == "pending"
|
|
|
|
|
|
def test_signal_today_endpoint(client: TestClient, auth_headers):
|
|
"""Test today's signals endpoint."""
|
|
response = client.get("/api/signal/kjb/today", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
assert isinstance(response.json(), list)
|
|
|
|
|
|
def test_signal_history_endpoint(client: TestClient, auth_headers):
|
|
"""Test signal history endpoint."""
|
|
response = client.get(
|
|
"/api/signal/kjb/history",
|
|
params={"limit": 10},
|
|
headers=auth_headers,
|
|
)
|
|
assert response.status_code == 200
|
|
assert isinstance(response.json(), list)
|
|
|
|
|
|
def test_signal_requires_auth(client: TestClient):
|
|
"""Test that signal endpoints require authentication."""
|
|
response = client.get("/api/signal/kjb/today")
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_cancel_executed_signal(client: TestClient, auth_headers):
|
|
"""실행된 신호를 취소하면 거래가 삭제되고 보유량이 복원된다."""
|
|
# 1. 포트폴리오 생성
|
|
resp = client.post(
|
|
"/api/portfolios",
|
|
json={"name": "Signal Cancel Test", "portfolio_type": "general"},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 201
|
|
portfolio_id = resp.json()["id"]
|
|
|
|
# 2. 신호 생성 (직접 DB 경유 없이 API로 생성 불가 → 신호 실행 취소는 EXECUTED 신호에만 작동)
|
|
# 오늘 날짜로 신호 조회해서 없으면 스킵
|
|
today_resp = client.get("/api/signal/kjb/today", headers=auth_headers)
|
|
signals = today_resp.json()
|
|
|
|
if not signals:
|
|
# 신호가 없으면 엔드포인트 존재만 검증 (portfolio_id 포함)
|
|
resp = client.delete("/api/signal/9999/cancel", params={"portfolio_id": 9999}, headers=auth_headers)
|
|
assert resp.status_code in [404, 400]
|
|
return
|
|
|
|
# 3. ACTIVE 신호에 보유 종목 세팅 후 신호 실행
|
|
signal = signals[0]
|
|
ticker = signal["ticker"]
|
|
|
|
client.put(
|
|
f"/api/portfolios/{portfolio_id}/holdings",
|
|
json=[{"ticker": ticker, "quantity": 100, "avg_price": 10000}],
|
|
headers=auth_headers,
|
|
)
|
|
|
|
exec_resp = client.post(
|
|
f"/api/signal/{signal['id']}/execute",
|
|
json={"portfolio_id": portfolio_id, "quantity": 10, "price": 10000},
|
|
headers=auth_headers,
|
|
)
|
|
# 신호 타입에 따라 실패할 수 있음
|
|
if exec_resp.status_code != 200:
|
|
return
|
|
|
|
# 4. 취소 요청
|
|
cancel_resp = client.delete(
|
|
f"/api/signal/{signal['id']}/cancel",
|
|
params={"portfolio_id": portfolio_id},
|
|
headers=auth_headers,
|
|
)
|
|
assert cancel_resp.status_code == 200
|
|
data = cancel_resp.json()
|
|
assert data["signal_status"] == "active"
|
|
assert data["transaction_deleted"] is True
|