d19e5b0de6
Some checks failed
CI / test (pull_request) Has been cancelled
스캐너 후보 종목뿐 아니라 현재 보유 종목도 매 사이클마다 평가해 stop-loss / take-profit이 실제로 동작하도록 개선. - db.py: get_open_positions_by_market() 추가 - net BUY - SELL 집계 쿼리로 실제 보유 종목 코드 목록 반환 - 단순 "최신 레코드 = BUY" 방식보다 안전 (이중 매도 방지) - main.py: 실시간 루프에서 스캐너 후보 + 보유 종목을 union으로 구성 - dict.fromkeys로 순서 유지하며 중복 제거 - 스캐너에 없는 보유 종목은 로그로 명시 - 보유 종목은 Playbook 없으면 HOLD → stop-loss/take-profit 체크 - tests/test_db.py: get_open_positions_by_market 테스트 5개 추가 - net 양수 종목 포함, 전량 매도 제외, 부분 매도 포함 - 마켓 범위 격리, 거래 없을 때 빈 리스트 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
145 lines
4.5 KiB
Python
145 lines
4.5 KiB
Python
"""Tests for database helper functions."""
|
|
|
|
from src.db import get_open_position, get_open_positions_by_market, init_db, log_trade
|
|
|
|
|
|
def test_get_open_position_returns_latest_buy() -> None:
|
|
conn = init_db(":memory:")
|
|
log_trade(
|
|
conn=conn,
|
|
stock_code="005930",
|
|
action="BUY",
|
|
confidence=90,
|
|
rationale="entry",
|
|
quantity=2,
|
|
price=70000.0,
|
|
market="KR",
|
|
exchange_code="KRX",
|
|
decision_id="d-buy-1",
|
|
)
|
|
|
|
position = get_open_position(conn, "005930", "KR")
|
|
assert position is not None
|
|
assert position["decision_id"] == "d-buy-1"
|
|
assert position["price"] == 70000.0
|
|
assert position["quantity"] == 2
|
|
|
|
|
|
def test_get_open_position_returns_none_when_latest_is_sell() -> None:
|
|
conn = init_db(":memory:")
|
|
log_trade(
|
|
conn=conn,
|
|
stock_code="005930",
|
|
action="BUY",
|
|
confidence=90,
|
|
rationale="entry",
|
|
quantity=1,
|
|
price=70000.0,
|
|
market="KR",
|
|
exchange_code="KRX",
|
|
decision_id="d-buy-1",
|
|
)
|
|
log_trade(
|
|
conn=conn,
|
|
stock_code="005930",
|
|
action="SELL",
|
|
confidence=95,
|
|
rationale="exit",
|
|
quantity=1,
|
|
price=71000.0,
|
|
market="KR",
|
|
exchange_code="KRX",
|
|
decision_id="d-sell-1",
|
|
)
|
|
|
|
assert get_open_position(conn, "005930", "KR") is None
|
|
|
|
|
|
def test_get_open_position_returns_none_when_no_trades() -> None:
|
|
conn = init_db(":memory:")
|
|
assert get_open_position(conn, "AAPL", "US_NASDAQ") is None
|
|
|
|
|
|
# --- get_open_positions_by_market tests ---
|
|
|
|
|
|
def test_get_open_positions_by_market_returns_net_positive_stocks() -> None:
|
|
"""Stocks with net BUY quantity > 0 are included."""
|
|
conn = init_db(":memory:")
|
|
log_trade(
|
|
conn=conn, stock_code="005930", action="BUY", confidence=90,
|
|
rationale="entry", quantity=5, price=70000.0, market="KR",
|
|
exchange_code="KRX", decision_id="d1",
|
|
)
|
|
log_trade(
|
|
conn=conn, stock_code="000660", action="BUY", confidence=85,
|
|
rationale="entry", quantity=3, price=100000.0, market="KR",
|
|
exchange_code="KRX", decision_id="d2",
|
|
)
|
|
|
|
result = get_open_positions_by_market(conn, "KR")
|
|
assert set(result) == {"005930", "000660"}
|
|
|
|
|
|
def test_get_open_positions_by_market_excludes_fully_sold_stocks() -> None:
|
|
"""Stocks where BUY qty == SELL qty are excluded (net qty = 0)."""
|
|
conn = init_db(":memory:")
|
|
log_trade(
|
|
conn=conn, stock_code="005930", action="BUY", confidence=90,
|
|
rationale="entry", quantity=3, price=70000.0, market="KR",
|
|
exchange_code="KRX", decision_id="d1",
|
|
)
|
|
log_trade(
|
|
conn=conn, stock_code="005930", action="SELL", confidence=95,
|
|
rationale="exit", quantity=3, price=71000.0, market="KR",
|
|
exchange_code="KRX", decision_id="d2",
|
|
)
|
|
|
|
result = get_open_positions_by_market(conn, "KR")
|
|
assert "005930" not in result
|
|
|
|
|
|
def test_get_open_positions_by_market_includes_partially_sold_stocks() -> None:
|
|
"""Stocks with partial SELL (net qty > 0) are still included."""
|
|
conn = init_db(":memory:")
|
|
log_trade(
|
|
conn=conn, stock_code="005930", action="BUY", confidence=90,
|
|
rationale="entry", quantity=5, price=70000.0, market="KR",
|
|
exchange_code="KRX", decision_id="d1",
|
|
)
|
|
log_trade(
|
|
conn=conn, stock_code="005930", action="SELL", confidence=95,
|
|
rationale="partial exit", quantity=2, price=71000.0, market="KR",
|
|
exchange_code="KRX", decision_id="d2",
|
|
)
|
|
|
|
result = get_open_positions_by_market(conn, "KR")
|
|
assert "005930" in result
|
|
|
|
|
|
def test_get_open_positions_by_market_is_market_scoped() -> None:
|
|
"""Only stocks from the specified market are returned."""
|
|
conn = init_db(":memory:")
|
|
log_trade(
|
|
conn=conn, stock_code="005930", action="BUY", confidence=90,
|
|
rationale="entry", quantity=3, price=70000.0, market="KR",
|
|
exchange_code="KRX", decision_id="d1",
|
|
)
|
|
log_trade(
|
|
conn=conn, stock_code="AAPL", action="BUY", confidence=85,
|
|
rationale="entry", quantity=2, price=200.0, market="NASD",
|
|
exchange_code="NAS", decision_id="d2",
|
|
)
|
|
|
|
kr_result = get_open_positions_by_market(conn, "KR")
|
|
nasd_result = get_open_positions_by_market(conn, "NASD")
|
|
|
|
assert kr_result == ["005930"]
|
|
assert nasd_result == ["AAPL"]
|
|
|
|
|
|
def test_get_open_positions_by_market_returns_empty_when_no_trades() -> None:
|
|
"""Empty list returned when no trades exist for the market."""
|
|
conn = init_db(":memory:")
|
|
assert get_open_positions_by_market(conn, "KR") == []
|