125 lines
4.1 KiB
Python
125 lines
4.1 KiB
Python
import os
|
|
import sqlite3
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.screener.schema import ensure_screener_schema
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def isolated_db(tmp_path, monkeypatch):
|
|
db_path = tmp_path / "screener_router.db"
|
|
c = sqlite3.connect(db_path)
|
|
ensure_screener_schema(c)
|
|
c.close()
|
|
monkeypatch.setenv("STOCK_DB_PATH", str(db_path))
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
from app.main import app
|
|
return TestClient(app)
|
|
|
|
|
|
def test_get_nodes_lists_7_score_and_1_gate(client):
|
|
r = client.get("/api/stock/screener/nodes")
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert len(body["score_nodes"]) == 7
|
|
assert len(body["gate_nodes"]) == 1
|
|
assert {n["name"] for n in body["score_nodes"]} == {
|
|
"foreign_buy", "volume_surge", "momentum",
|
|
"high52w", "rs_rating", "ma_alignment", "vcp_lite",
|
|
}
|
|
|
|
|
|
def test_settings_get_returns_defaults(client):
|
|
r = client.get("/api/stock/screener/settings")
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["weights"]["foreign_buy"] == 1.0
|
|
assert body["top_n"] == 20
|
|
|
|
|
|
def test_settings_put_then_get_round_trip(client):
|
|
new_settings = {
|
|
"weights": {"foreign_buy": 2.5, "momentum": 1.0, "volume_surge": 1.0,
|
|
"high52w": 1.2, "rs_rating": 1.2, "ma_alignment": 1.0, "vcp_lite": 0.8},
|
|
"node_params": {"foreign_buy": {"window_days": 7}},
|
|
"gate_params": {"min_market_cap_won": 100_000_000_000,
|
|
"min_avg_value_won": 500_000_000,
|
|
"min_listed_days": 60,
|
|
"skip_managed": True, "skip_preferred": True, "skip_spac": True,
|
|
"skip_halted_days": 3},
|
|
"top_n": 30,
|
|
"rr_ratio": 2.5,
|
|
"atr_window": 14,
|
|
"atr_stop_mult": 2.0,
|
|
}
|
|
r = client.put("/api/stock/screener/settings", json=new_settings)
|
|
assert r.status_code == 200
|
|
r2 = client.get("/api/stock/screener/settings")
|
|
body = r2.json()
|
|
assert body["weights"]["foreign_buy"] == 2.5
|
|
assert body["top_n"] == 30
|
|
|
|
|
|
# ---- /run tests ----
|
|
|
|
from app.screener._test_fixtures import make_master, make_prices, make_flow
|
|
|
|
|
|
def _seed_min(conn, asof_iso="2026-05-12"):
|
|
import datetime as dt
|
|
now = dt.datetime.utcnow().isoformat()
|
|
rows = [
|
|
("BIG1", "큰주식1", "KOSPI", 200_000_000_000, 0, 0, 0, None, now),
|
|
("BIG2", "큰주식2", "KOSPI", 100_000_000_000, 0, 0, 0, None, now),
|
|
("SMALL", "작은주식", "KOSPI", 1_000_000_000, 0, 0, 0, None, now),
|
|
]
|
|
for r in rows:
|
|
conn.execute("""INSERT INTO krx_master (ticker,name,market,market_cap,
|
|
is_managed,is_preferred,is_spac,listed_date,updated_at)
|
|
VALUES (?,?,?,?,?,?,?,?,?)""", r)
|
|
asof = dt.date(2026, 5, 12)
|
|
p = make_prices(["BIG1", "BIG2", "SMALL"], days=260, asof=asof)
|
|
f = make_flow(["BIG1", "BIG2", "SMALL"], days=260, asof=asof,
|
|
foreign_per_day={"BIG1": 100_000_000, "BIG2": 50_000_000, "SMALL": 0})
|
|
p.to_sql("krx_daily_prices", conn, if_exists="append", index=False)
|
|
f.to_sql("krx_flow", conn, if_exists="append", index=False)
|
|
conn.commit()
|
|
|
|
|
|
def test_run_preview_no_save(client):
|
|
db_path = os.environ["STOCK_DB_PATH"]
|
|
c = sqlite3.connect(db_path)
|
|
_seed_min(c)
|
|
c.close()
|
|
|
|
r = client.post("/api/stock/screener/run", json={"mode": "preview", "asof": "2026-05-12"})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["status"] == "success"
|
|
assert body["run_id"] is None
|
|
assert body["telegram_payload"] is not None
|
|
|
|
c = sqlite3.connect(db_path)
|
|
cnt = c.execute("SELECT count(*) FROM screener_runs").fetchone()[0]
|
|
assert cnt == 0
|
|
|
|
|
|
def test_run_manual_save_writes_row(client):
|
|
db_path = os.environ["STOCK_DB_PATH"]
|
|
c = sqlite3.connect(db_path)
|
|
_seed_min(c)
|
|
c.close()
|
|
|
|
r = client.post("/api/stock/screener/run",
|
|
json={"mode": "manual_save", "asof": "2026-05-12"})
|
|
assert r.status_code == 200
|
|
assert r.json()["run_id"] is not None
|
|
|
|
c = sqlite3.connect(db_path)
|
|
cnt = c.execute("SELECT count(*) FROM screener_runs").fetchone()[0]
|
|
assert cnt == 1
|