102 lines
4.0 KiB
Python
102 lines
4.0 KiB
Python
import sqlite3
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def conn(monkeypatch, tmp_path):
|
|
from app import db as _db
|
|
monkeypatch.setattr(_db, "DB_PATH", str(tmp_path / "stock.db"))
|
|
_db.init_db()
|
|
c = sqlite3.connect(_db.DB_PATH)
|
|
c.row_factory = sqlite3.Row
|
|
# 보유 1종목 (add_portfolio_item 실제 시그니처: broker/ticker/name/quantity/avg_price — market 파라미터 없음)
|
|
_db.add_portfolio_item(ticker="000660", name="SK하이닉스", quantity=10,
|
|
avg_price=180000, broker="kis")
|
|
# watchlist 1종목
|
|
_db.add_watchlist("005930", "삼성전자")
|
|
yield c
|
|
c.close()
|
|
|
|
|
|
def test_build_monitor_set_merges_sources(conn):
|
|
from app import trade_alerts as ta
|
|
ms = ta.build_monitor_set(conn, session="regular",
|
|
exit_params={"stop_pct": 0.08}, buy_params={"rsi_oversold": 30})
|
|
buy_tickers = {t["ticker"] for t in ms["buy_targets"]}
|
|
sell_tickers = {t["ticker"] for t in ms["sell_targets"]}
|
|
assert "005930" in buy_tickers # watchlist
|
|
assert "000660" in sell_tickers # 보유
|
|
assert ms["session"] == "regular"
|
|
assert ms["exit_params"]["stop_pct"] == 0.08
|
|
sell = next(t for t in ms["sell_targets"] if t["ticker"] == "000660")
|
|
assert sell["avg_price"] == 180000 and sell["qty"] == 10
|
|
|
|
|
|
def test_latest_screener_candidates_empty_when_no_run(conn):
|
|
from app import trade_alerts as ta
|
|
assert ta.latest_screener_candidates(conn) == []
|
|
|
|
|
|
def test_latest_screener_candidates_picks_latest_success_run(conn):
|
|
from app import trade_alerts as ta
|
|
now = "2026-07-02T09:00:00Z"
|
|
conn.execute(
|
|
"INSERT INTO screener_runs (asof, mode, status, started_at, weights_json, "
|
|
"node_params_json, gate_params_json, top_n) VALUES (?,?,?,?,?,?,?,?)",
|
|
(now, "manual", "failed", now, "{}", "{}", "{}", 20),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO screener_runs (asof, mode, status, started_at, weights_json, "
|
|
"node_params_json, gate_params_json, top_n) VALUES (?,?,?,?,?,?,?,?)",
|
|
(now, "manual", "success", now, "{}", "{}", "{}", 20),
|
|
)
|
|
run_id = conn.execute("SELECT id FROM screener_runs WHERE status='success'").fetchone()[0]
|
|
conn.execute(
|
|
"INSERT INTO screener_results (run_id, rank, ticker, name, total_score, scores_json) "
|
|
"VALUES (?,?,?,?,?,?)",
|
|
(run_id, 1, "035720", "카카오", 88.5, "{}"),
|
|
)
|
|
conn.commit()
|
|
candidates = ta.latest_screener_candidates(conn)
|
|
assert candidates == [{"ticker": "035720", "name": "카카오"}]
|
|
|
|
|
|
def test_holding_high_returns_max_high_within_lookback(conn):
|
|
from app import trade_alerts as ta
|
|
conn.execute(
|
|
"INSERT INTO krx_daily_prices (ticker, date, high) VALUES (?,?,?)",
|
|
("000660", "2026-06-01", 200000),
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO krx_daily_prices (ticker, date, high) VALUES (?,?,?)",
|
|
("000660", "2026-06-20", 210000),
|
|
)
|
|
conn.commit()
|
|
assert ta.holding_high(conn, "000660", lookback_days=60) == 210000
|
|
|
|
|
|
def test_holding_high_none_when_no_price_history(conn):
|
|
from app import trade_alerts as ta
|
|
assert ta.holding_high(conn, "999999") is None
|
|
|
|
|
|
def test_build_monitor_set_dedupes_watchlist_and_screener_overlap(conn):
|
|
from app import trade_alerts as ta
|
|
now = "2026-07-02T09:00:00Z"
|
|
cur = conn.execute(
|
|
"INSERT INTO screener_runs (asof, mode, status, started_at, weights_json, "
|
|
"node_params_json, gate_params_json, top_n) VALUES (?,?,?,?,?,?,?,?)",
|
|
(now, "manual", "success", now, "{}", "{}", "{}", 20),
|
|
)
|
|
run_id = cur.lastrowid
|
|
# 스크리너 후보가 watchlist와 중복(005930)
|
|
conn.execute(
|
|
"INSERT INTO screener_results (run_id, rank, ticker, name, total_score, scores_json) "
|
|
"VALUES (?,?,?,?,?,?)",
|
|
(run_id, 1, "005930", "삼성전자", 90.0, "{}"),
|
|
)
|
|
conn.commit()
|
|
ms = ta.build_monitor_set(conn, session="regular", exit_params={}, buy_params={})
|
|
buy_tickers = [t["ticker"] for t in ms["buy_targets"]]
|
|
assert buy_tickers.count("005930") == 1
|