gahusb
5f0fed7f13
- screener/router.py: APIRouter prefix=/api/stock/screener - GET /nodes: NODE_REGISTRY + GATE_REGISTRY 메타 노출 (7 score + 1 gate) - GET /settings: screener_settings 싱글톤 row 조회 - PUT /settings: 가중치/노드/게이트 파라미터 round-trip - main.py: screener_router include (FastAPI 생성 직후) - db.py: STOCK_DB_PATH 환경변수 지원 (테스트 격리, 기본값 /app/data/stock.db 유지) - test_screener_router.py: 3 tests (nodes list, settings GET, PUT round-trip)
86 lines
2.5 KiB
Python
86 lines
2.5 KiB
Python
"""FastAPI router for /api/stock/screener/*"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
|
|
from . import schemas
|
|
from .registry import NODE_REGISTRY, GATE_REGISTRY
|
|
|
|
|
|
router = APIRouter(prefix="/api/stock/screener")
|
|
|
|
|
|
def _db_path() -> str:
|
|
return os.environ.get("STOCK_DB_PATH", "/app/data/stock.db")
|
|
|
|
|
|
def _conn() -> sqlite3.Connection:
|
|
return sqlite3.connect(_db_path())
|
|
|
|
|
|
# ---------- /nodes ----------
|
|
|
|
@router.get("/nodes", response_model=schemas.NodesResponse)
|
|
def get_nodes():
|
|
score_nodes = [
|
|
schemas.NodeMeta(
|
|
name=cls.name, label=cls.label,
|
|
default_params=cls.default_params, param_schema=cls.param_schema,
|
|
)
|
|
for cls in NODE_REGISTRY.values()
|
|
]
|
|
gate_nodes = [
|
|
schemas.NodeMeta(
|
|
name=cls.name, label=cls.label,
|
|
default_params=cls.default_params, param_schema=cls.param_schema,
|
|
)
|
|
for cls in GATE_REGISTRY.values()
|
|
]
|
|
return schemas.NodesResponse(score_nodes=score_nodes, gate_nodes=gate_nodes)
|
|
|
|
|
|
# ---------- /settings ----------
|
|
|
|
@router.get("/settings", response_model=schemas.SettingsResponse)
|
|
def get_settings():
|
|
with _conn() as c:
|
|
row = c.execute(
|
|
"SELECT weights_json, node_params_json, gate_params_json, "
|
|
"top_n, rr_ratio, atr_window, atr_stop_mult, updated_at "
|
|
"FROM screener_settings WHERE id=1"
|
|
).fetchone()
|
|
if row is None:
|
|
raise HTTPException(503, "settings not initialized")
|
|
return schemas.SettingsResponse(
|
|
weights=json.loads(row[0]),
|
|
node_params=json.loads(row[1]),
|
|
gate_params=json.loads(row[2]),
|
|
top_n=row[3], rr_ratio=row[4], atr_window=row[5], atr_stop_mult=row[6],
|
|
updated_at=row[7],
|
|
)
|
|
|
|
|
|
@router.put("/settings", response_model=schemas.SettingsResponse)
|
|
def put_settings(body: schemas.SettingsBody):
|
|
now = dt.datetime.utcnow().isoformat()
|
|
with _conn() as c:
|
|
c.execute(
|
|
"""UPDATE screener_settings SET
|
|
weights_json=?, node_params_json=?, gate_params_json=?,
|
|
top_n=?, rr_ratio=?, atr_window=?, atr_stop_mult=?, updated_at=?
|
|
WHERE id=1""",
|
|
(
|
|
json.dumps(body.weights), json.dumps(body.node_params),
|
|
json.dumps(body.gate_params),
|
|
body.top_n, body.rr_ratio, body.atr_window, body.atr_stop_mult, now,
|
|
),
|
|
)
|
|
c.commit()
|
|
return schemas.SettingsResponse(**body.model_dump(), updated_at=now)
|