feat(stock-lab): ScoreNode/GateNode 추상 + percentile_rank 유틸
This commit is contained in:
0
stock-lab/app/screener/nodes/__init__.py
Normal file
0
stock-lab/app/screener/nodes/__init__.py
Normal file
40
stock-lab/app/screener/nodes/base.py
Normal file
40
stock-lab/app/screener/nodes/base.py
Normal file
@@ -0,0 +1,40 @@
|
||||
"""Node base classes + helpers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Any, ClassVar
|
||||
|
||||
import pandas as pd
|
||||
|
||||
|
||||
class ScoreNode(ABC):
|
||||
name: ClassVar[str]
|
||||
label: ClassVar[str]
|
||||
default_params: ClassVar[dict]
|
||||
param_schema: ClassVar[dict]
|
||||
|
||||
@abstractmethod
|
||||
def compute(self, ctx: "Any", params: dict) -> pd.Series:
|
||||
"""returns Series indexed by ticker, 0..100 float."""
|
||||
|
||||
|
||||
class GateNode(ABC):
|
||||
name: ClassVar[str]
|
||||
label: ClassVar[str]
|
||||
default_params: ClassVar[dict]
|
||||
param_schema: ClassVar[dict]
|
||||
|
||||
@abstractmethod
|
||||
def filter(self, ctx: "Any", params: dict) -> pd.Index:
|
||||
"""returns surviving tickers."""
|
||||
|
||||
|
||||
def percentile_rank(series: pd.Series) -> pd.Series:
|
||||
"""Percentile rank in [0, 100]. All-equal → 50. NaN preserved."""
|
||||
if series.empty:
|
||||
return series.astype(float)
|
||||
if series.dropna().nunique() == 1:
|
||||
return pd.Series(50.0, index=series.index)
|
||||
ranked = series.rank(pct=True, na_option="keep") * 100.0
|
||||
return ranked
|
||||
24
stock-lab/app/test_screener_nodes_base.py
Normal file
24
stock-lab/app/test_screener_nodes_base.py
Normal file
@@ -0,0 +1,24 @@
|
||||
import pandas as pd
|
||||
import pytest
|
||||
|
||||
from app.screener.nodes.base import percentile_rank
|
||||
|
||||
|
||||
def test_percentile_rank_basic():
|
||||
s = pd.Series([10, 20, 30, 40, 50])
|
||||
out = percentile_rank(s)
|
||||
assert (out >= 0).all() and (out <= 100).all()
|
||||
assert out.iloc[0] < out.iloc[-1] # smallest gets lowest rank
|
||||
|
||||
|
||||
def test_percentile_rank_all_equal_returns_50():
|
||||
s = pd.Series([42, 42, 42, 42])
|
||||
out = percentile_rank(s)
|
||||
assert (out == 50.0).all()
|
||||
|
||||
|
||||
def test_percentile_rank_handles_nan():
|
||||
s = pd.Series([1.0, float("nan"), 3.0, 5.0])
|
||||
out = percentile_rank(s)
|
||||
assert pd.isna(out.iloc[1])
|
||||
assert (out.dropna() >= 0).all()
|
||||
Reference in New Issue
Block a user