130 lines
4.4 KiB
Python
130 lines
4.4 KiB
Python
import datetime as dt
|
|
import sqlite3
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from app.screener import snapshot as snap
|
|
from app.screener.schema import ensure_screener_schema
|
|
|
|
|
|
@pytest.fixture
|
|
def conn(tmp_path):
|
|
db_path = tmp_path / "snap.db"
|
|
c = sqlite3.connect(db_path)
|
|
ensure_screener_schema(c)
|
|
yield c
|
|
c.close()
|
|
|
|
|
|
def _stub_listing(monkeypatch):
|
|
df = pd.DataFrame([
|
|
{"Code": "005930", "Name": "삼성전자", "Market": "KOSPI",
|
|
"Marcap": 420_000_000_000_000,
|
|
"Open": 70000, "High": 72000, "Low": 69500, "Close": 71000,
|
|
"Volume": 12_000_000, "Amount": 840_000_000_000},
|
|
{"Code": "035420", "Name": "NAVER", "Market": "KOSPI",
|
|
"Marcap": 30_000_000_000_000,
|
|
"Open": 215000, "High": 220000, "Low": 213000, "Close": 218000,
|
|
"Volume": 1_000_000, "Amount": 218_000_000_000},
|
|
{"Code": "091990", "Name": "셀트리온헬스케어우", "Market": "KOSDAQ",
|
|
"Marcap": 10_000_000_000_000,
|
|
"Open": 60000, "High": 61000, "Low": 59500, "Close": 60500,
|
|
"Volume": 500_000, "Amount": 30_250_000_000},
|
|
])
|
|
monkeypatch.setattr(snap, "fetch_master_listing", lambda: df)
|
|
|
|
|
|
def _stub_flow(monkeypatch, mapping):
|
|
def fake_flow(ticker, *, client):
|
|
if mapping is None:
|
|
return None
|
|
v = mapping.get(ticker)
|
|
if v is None:
|
|
return None
|
|
return {
|
|
"date": dt.date(2026, 5, 12).isoformat(),
|
|
"foreign_net": v["foreign_net"],
|
|
"institution_net": v["institution_net"],
|
|
}
|
|
monkeypatch.setattr(snap, "fetch_flow_naver", fake_flow)
|
|
|
|
|
|
def test_refresh_daily_writes_master_and_prices(conn, monkeypatch):
|
|
_stub_listing(monkeypatch)
|
|
_stub_flow(monkeypatch, None)
|
|
summary = snap.refresh_daily(conn, dt.date(2026, 5, 12),
|
|
flow_top_n=10, rate_limit_sec=0)
|
|
assert summary["master_count"] == 3
|
|
assert summary["prices_count"] == 3
|
|
assert summary["flow_count"] == 0
|
|
|
|
row = conn.execute(
|
|
"SELECT close FROM krx_daily_prices WHERE ticker='005930' AND date='2026-05-12'"
|
|
).fetchone()
|
|
assert row[0] == 71000
|
|
|
|
|
|
def test_refresh_daily_writes_flow_for_top_n(conn, monkeypatch):
|
|
_stub_listing(monkeypatch)
|
|
_stub_flow(monkeypatch, {
|
|
"005930": {"foreign_net": 12_000_000_000, "institution_net": 4_000_000_000},
|
|
"035420": {"foreign_net": -3_000_000_000, "institution_net": 8_000_000_000},
|
|
})
|
|
summary = snap.refresh_daily(conn, dt.date(2026, 5, 12),
|
|
flow_top_n=2, rate_limit_sec=0)
|
|
assert summary["flow_count"] == 2
|
|
row = conn.execute(
|
|
"SELECT foreign_net FROM krx_flow WHERE ticker='005930'"
|
|
).fetchone()
|
|
assert row[0] == 12_000_000_000
|
|
|
|
|
|
def test_master_flags_preferred(conn, monkeypatch):
|
|
_stub_listing(monkeypatch)
|
|
_stub_flow(monkeypatch, None)
|
|
snap.refresh_daily(conn, dt.date(2026, 5, 12), flow_top_n=0, rate_limit_sec=0)
|
|
pref = conn.execute(
|
|
"SELECT is_preferred FROM krx_master WHERE ticker='091990'"
|
|
).fetchone()
|
|
assert pref[0] == 1
|
|
|
|
|
|
def test_refresh_daily_is_idempotent(conn, monkeypatch):
|
|
_stub_listing(monkeypatch)
|
|
_stub_flow(monkeypatch, None)
|
|
snap.refresh_daily(conn, dt.date(2026, 5, 12), flow_top_n=0, rate_limit_sec=0)
|
|
snap.refresh_daily(conn, dt.date(2026, 5, 12), flow_top_n=0, rate_limit_sec=0)
|
|
cnt = conn.execute(
|
|
"SELECT count(*) FROM krx_daily_prices WHERE date='2026-05-12'"
|
|
).fetchone()[0]
|
|
assert cnt == 3
|
|
|
|
|
|
def test_fetch_flow_naver_parses_html():
|
|
"""Real HTML structure parse with synthetic naver-like markup."""
|
|
html = """
|
|
<html><body>
|
|
<table class="type2">
|
|
<tr><th>날짜</th></tr>
|
|
<tr><td>2026.05.12</td><td>71,000</td><td>500</td><td>0.71%</td>
|
|
<td>12,000,000</td><td>4,000,000,000</td><td>12,000,000,000</td>
|
|
<td>1</td><td>53.0</td></tr>
|
|
<tr><td>2026.05.09</td><td>70,500</td><td>-200</td><td>-0.28%</td>
|
|
<td>10,000,000</td><td>2,000,000,000</td><td>5,000,000,000</td>
|
|
<td>1</td><td>52.8</td></tr>
|
|
</table>
|
|
</body></html>
|
|
"""
|
|
class FakeResp:
|
|
status_code = 200
|
|
text = html
|
|
class FakeClient:
|
|
def get(self, url, params): return FakeResp()
|
|
out = snap.fetch_flow_naver("005930", client=FakeClient())
|
|
assert out == {
|
|
"date": "2026-05-12",
|
|
"foreign_net": 12_000_000_000,
|
|
"institution_net": 4_000_000_000,
|
|
}
|