This commit is contained in:
Nolan Kovacs 2026-04-04 09:49:22 -04:00
parent f2c8daefba
commit e17d63ceeb
12 changed files with 4479 additions and 5160 deletions

View File

@ -10,7 +10,25 @@
"Bash(ls /c/Python*)", "Bash(ls /c/Python*)",
"Read(//c/Users/Nolan/AppData/Local/Programs/**)", "Read(//c/Users/Nolan/AppData/Local/Programs/**)",
"Bash(ls \"/c/Users/Nolan/AppData/Local/Programs/Python\" 2>/dev/null || echo \"not found\"\nls \"/c/Users/Nolan/AppData/Local/Microsoft/WindowsApps/\"python* 2>/dev/null || echo \"not found\")", "Bash(ls \"/c/Users/Nolan/AppData/Local/Programs/Python\" 2>/dev/null || echo \"not found\"\nls \"/c/Users/Nolan/AppData/Local/Microsoft/WindowsApps/\"python* 2>/dev/null || echo \"not found\")",
"Bash(cd \"C:/Users/Nolan/Desktop/Stock Tool\" && python -c \"import tkinter; import numpy; import pandas; import yfinance; import requests; print\\('All imports OK'\\)\" 2>&1)" "Bash(cd \"C:/Users/Nolan/Desktop/Stock Tool\" && python -c \"import tkinter; import numpy; import pandas; import yfinance; import requests; print\\('All imports OK'\\)\" 2>&1)",
"Skill(update-config)"
]
},
"hooks": {
"PostToolUse": [
{
"matcher": "Edit|Write",
"hooks": [
{
"type": "command",
"command": "cd \"C:/Users/noach/Desktop/Stock Tool/Stock Tool\" && python -m py_compile dist/stock_screener.py dist/screener_gui.py cache_builder.py api/main.py 2>&1 && echo \"[hook] SYNTAX OK\" || echo \"[hook] SYNTAX ERROR — see above\""
},
{
"type": "command",
"command": "cd \"C:/Users/noach/Desktop/Stock Tool/Stock Tool\" && python -m pytest tests/test_scoring.py -q --tb=short 2>&1 | tail -20"
}
]
}
] ]
} }
} }

108
CLAUDE.md Normal file
View File

@ -0,0 +1,108 @@
# Stock Tool — Claude Context
## Project Overview
A multi-factor stock screener that evaluates ~6,500 US-listed equities using a composite
score across 8 dimensions: Value, Growth, Momentum, Quality, Profitability, Sentiment,
Analyst, and Risk. Sold as a licensed desktop app (PySide6 GUI).
## File Map
| File | Purpose |
|------|---------|
| `dist/stock_screener.py` | All scoring logic, data fetching, index/ticker collection. Pure Python + pandas. |
| `dist/screener_gui.py` | PySide6 GUI. Imports from `stock_screener`. Main class: `ScreenerApp(QMainWindow)`. |
| `cache_builder.py` | Nightly job that populates PostgreSQL cache from EDGAR + yfinance. Runs on VPS via cron. |
| `api/main.py` | FastAPI server on VPS. Handles license auth, cache serving, version management. |
| `updater.py` | Client-side auto-updater. Checks `/update` endpoint, downloads new build. |
| `license_check.py` | HMAC + JWT license validation. Baked into app via PyArmor. |
| `push_update.py` | Dev tool: SCP new build to VPS + register version via `/admin/register-version`. |
| `launcher.py` | Thin launcher that checks license before starting GUI. |
| `health_monitor.py` | VPS health monitoring. |
| `discord_bot/bot.py` | Discord bot integration (separate process). |
## Architecture (3-tier)
```
[cache_builder.py] → [PostgreSQL on VPS] → [API] → [Client app]
nightly 6 tables FastAPI screener_gui + stock_screener
```
- `cache_builder.py` runs nightly (2am stable / 3am beta) via cron on the VPS
- Data flows: EDGAR XBRL Frames API + yfinance → PostgreSQL tables → served by FastAPI
- Client fetches cached data at startup via `/cache/analyst`, `/cache/fundamentals`, `/cache/sector`
- Scoring runs entirely client-side in `stock_screener.py` — no server-side scoring
## PostgreSQL Tables (database: `verimund`, user: `verimund_user`)
| Table | Contents |
|-------|----------|
| `analyst_cache` | Price history, returns, sentiment, analyst ratings, short interest (~6,975 rows) |
| `fundamentals_cache` | EDGAR-sourced financials: revenue, margins, D/E, ROA/ROE, Piotroski, accruals |
| `sector_stats` | Per-sector median/MAD for each metric. Used by `_z_score()` for normalization. |
| `app_versions` | Version registry per channel (stable/beta). Read by `/admin/versions`. |
| `licenses` | License keys + HWIDs + expiry. Validated at auth time. |
| `auth_log` | Auth attempt log. |
## Scoring Formula
```
Composite = 0.18*Value + 0.18*Growth + 0.14*Momentum + 0.11*Quality
+ 0.08*Profitability + 0.13*Sentiment + 0.10*Analyst + 0.08*Risk
```
Each score is 0100. Z-scores use sector median/MAD from `sector_stats` table.
If sector stats are missing for a metric, `_z_score()` returns neutral 50.
## Profile System (important)
`_classify_profile(row)` classifies each stock into one of:
`mature`, `high_growth`, `financial`, `capital_intensive`, `early_stage`, `turnaround`
Profiles define sub-weights per scoring dimension in `_PROFILE_WEIGHTS`.
Weights are normalized in `_profile_score_row()` — they don't need to sum to 1.0 in the dict.
**D/E handling**: yfinance returns `debtToEquity` as percentage (e.g. 150 = 1.5x).
Code divides by 100 at ingest (`stock_screener.py` line ~1356).
Negative D/E means negative book equity — replaced with 999.0 before z-scoring.
## Key Patterns
- `_z_series(series, metric, sectors, sector_stats)` — vectorized z-score using sector stats
- `_score_series(series, breakpoints)` — absolute breakpoint scoring (no sector normalization)
- `_profile_score_row(row, dimension, metric_scores)` — blend metric scores using profile weights
- All scoring functions signature: `compute_X_score(df, sector_stats=None) → pd.Series`
## VPS Infrastructure
- **Provider**: Hetzner CPX21, Ubuntu 22.04
- **Domains**: api.verimundsolutions.com (API), git.verimundsolutions.com (Gitea)
- **API**: FastAPI via uvicorn, managed by PM2
- **Secrets**: `/srv/api/.env` on VPS (DB_PASS, JWT_SECRET, HMAC_SECRET, ADMIN_KEY)
- **Backups**: nightly rclone → Cloudflare R2 bucket `verimund-backups`
- **Git**: Gitea at git.verimundsolutions.com, repo: ssz223/Stock-Tool, branches: main (stable) / beta
- **Website**: Taken down intentionally (2026-04-03), will be re-done under different brand
## Running Tests
```bash
# Fast tests (run automatically via hook on every edit):
python -m pytest tests/test_scoring.py tests/test_gui.py -q
# Full suite including network (run before pushing):
python -m pytest tests/ -v
# Network tests only:
python -m pytest tests/test_network.py -v
# Skip network tests:
python -m pytest tests/ -m "not network" -v
```
## Known Pitfalls
- `screener_gui.py` uses PySide6 (not PyQt5). Offscreen: `QT_QPA_PLATFORM=offscreen`
- `screener_gui.py` imports from `stock_screener` (same dir) — tests must `cd dist/` or add dist to sys.path
- EDGAR XBRL Frames API: one call per concept returns all US filers. Foreign/IFRS filers can have data 18 months old with no user-facing flag.
- Sector stats built from `analyst_cache` in Phase 5 of cache_builder. If Phase 1 partially fails, sector medians are biased.
- `_sum_frames` requires >= 4 quarters for valid TTM — returns None otherwise.
- Forward PE not in sector stats — `_z_score` returns neutral 50 for forward PE z-scoring.

View File

@ -20,7 +20,9 @@ app = FastAPI()
DB_HOST = "127.0.0.1" DB_HOST = "127.0.0.1"
DB_NAME = "verimund" DB_NAME = "verimund"
DB_USER = "verimund_user" DB_USER = "verimund_user"
DB_PASS = os.getenv("DB_PASS", "Shlevison2k17") DB_PASS = os.getenv("DB_PASS")
if not DB_PASS:
raise RuntimeError("DB_PASS environment variable is not set")
JWT_SECRET = os.getenv("JWT_SECRET", "") JWT_SECRET = os.getenv("JWT_SECRET", "")
HMAC_SECRET = os.getenv("HMAC_SECRET", "") HMAC_SECRET = os.getenv("HMAC_SECRET", "")
ADMIN_KEY = os.getenv("ADMIN_KEY", "") ADMIN_KEY = os.getenv("ADMIN_KEY", "")
@ -157,11 +159,14 @@ def check_update(payload=Depends(verify_jwt), db=Depends(get_db)):
@app.get("/admin/versions") @app.get("/admin/versions")
def get_versions(_=Depends(verify_admin), db=Depends(get_db)): def get_versions(channel: str = "stable", _=Depends(verify_admin), db=Depends(get_db)):
cur = db.cursor(cursor_factory=psycopg2.extras.RealDictCursor) cur = db.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
cur.execute("SELECT * FROM app_versions ORDER BY released_at DESC LIMIT 1") cur.execute(
"SELECT * FROM app_versions WHERE channel = %s ORDER BY released_at DESC LIMIT 1",
(channel,)
)
version = cur.fetchone() version = cur.fetchone()
return {"version": version["version"] if version else "1.0.0"} return {"version": version["version"] if version else "1.0.0", "channel": channel}
@app.post("/admin/register-version") @app.post("/admin/register-version")

View File

@ -474,7 +474,7 @@ _ANALYST_COLS = (
"news_headline_count", "sector", "industry", "short_percent", "short_ratio", "updated_at", "news_headline_count", "sector", "industry", "short_percent", "short_ratio", "updated_at",
) )
_FUND_COLS = ( _FUND_COLS = (
"ticker", "pe_trailing", "pb_ratio", "ev_ebitda", "eps_trailing", "ticker", "name", "pe_trailing", "pb_ratio", "ev_ebitda", "eps_trailing",
"revenue", "revenue_growth", "earnings_growth", "roe", "roa", "revenue", "revenue_growth", "earnings_growth", "roe", "roa",
"debt_to_equity", "total_debt", "total_cash", "book_value", "current_ratio", "debt_to_equity", "total_debt", "total_cash", "book_value", "current_ratio",
"profit_margin", "operating_margin", "fcf_yield", "dividend_yield", "profit_margin", "operating_margin", "fcf_yield", "dividend_yield",
@ -747,7 +747,13 @@ def run_fundamentals_phase(all_tickers: list[str]) -> dict[str, dict]:
cash.get(cik), lt_debt.get(cik), cash.get(cik), lt_debt.get(cik),
shares.get(cik), shares.get(cik),
) )
try:
_info = yf.Ticker(ticker).info or {}
_name = _info.get("longName") or _info.get("shortName") or None
except Exception:
_name = None
metrics["ticker"] = ticker metrics["ticker"] = ticker
metrics["name"] = _name
metrics["filer_type"] = "us-gaap-quarterly" metrics["filer_type"] = "us-gaap-quarterly"
metrics["updated_at"] = datetime.now(timezone.utc).isoformat() metrics["updated_at"] = datetime.now(timezone.utc).isoformat()
result[ticker] = metrics result[ticker] = metrics
@ -778,7 +784,13 @@ def run_fundamentals_phase(all_tickers: list[str]) -> dict[str, dict]:
cash.get(cik), lt_debt.get(cik), cash.get(cik), lt_debt.get(cik),
shares.get(cik), shares.get(cik),
) )
try:
_info = yf.Ticker(ticker).info or {}
_name = _info.get("longName") or _info.get("shortName") or None
except Exception:
_name = None
metrics["ticker"] = ticker metrics["ticker"] = ticker
metrics["name"] = _name
metrics["filer_type"] = "us-gaap-quarterly" metrics["filer_type"] = "us-gaap-quarterly"
metrics["updated_at"] = datetime.now(timezone.utc).isoformat() metrics["updated_at"] = datetime.now(timezone.utc).isoformat()
result[ticker] = metrics result[ticker] = metrics
@ -915,7 +927,13 @@ def run_foreign_filers_phase() -> dict[str, dict]:
raw.get("cash"), raw.get("lt_debt"), raw.get("cash"), raw.get("lt_debt"),
raw.get("shares"), raw.get("shares"),
) )
try:
_info = yf.Ticker(ticker).info or {}
_name = _info.get("longName") or _info.get("shortName") or None
except Exception:
_name = None
metrics["ticker"] = ticker metrics["ticker"] = ticker
metrics["name"] = _name
metrics["filer_type"] = f"{taxonomy}-annual" metrics["filer_type"] = f"{taxonomy}-annual"
metrics["updated_at"] = datetime.now(timezone.utc).isoformat() metrics["updated_at"] = datetime.now(timezone.utc).isoformat()
result[ticker] = metrics result[ticker] = metrics

7639
dist/screener_gui.py vendored

File diff suppressed because it is too large Load Diff

View File

@ -791,7 +791,8 @@ def _fetch_one_frame(concept: str, unit: str, period: str) -> dict[int, float]:
time.sleep(0.15) # stay comfortably under SEC's 10 req/sec limit time.sleep(0.15) # stay comfortably under SEC's 10 req/sec limit
return result return result
except Exception: except Exception:
_FRAME_CACHE[key] = {} # Do not cache transient failures (network errors, timeouts, etc.)
# Only 404s (already handled above) should produce a permanent empty cache entry.
return {} return {}
@ -808,7 +809,7 @@ def _sum_frames(concepts: list[str], unit: str,
ttm.setdefault(cik, {}) ttm.setdefault(cik, {})
if period not in ttm[cik]: if period not in ttm[cik]:
ttm[cik][period] = val ttm[cik][period] = val
return {cik: sum(pv.values()) for cik, pv in ttm.items()} return {cik: sum(pv.values()) for cik, pv in ttm.items() if len(pv) >= 4}
def _best_frame(concepts: list[str], unit: str, def _best_frame(concepts: list[str], unit: str,
@ -1591,8 +1592,11 @@ def fetch_all(
"analyst_count": analyst.get("analyst_count"), "analyst_count": analyst.get("analyst_count"),
"analyst_target": analyst_target, "analyst_target": analyst_target,
"recommendation": analyst.get("recommendation", "N/A"), "recommendation": analyst.get("recommendation", "N/A"),
# yfinance shortPercentOfFloat (% of float) preferred over FINRA short volume ratio # Bug 10 fix: use only shortPercentOfFloat (0.01.0 float fraction) from yfinance.
"short_percent": analyst.get("short_percent") or finra_short.get(ticker), # FINRA short-volume ratio (sv/total_vol) is on a different scale (typically 0.30.6)
# and is incompatible with the short_pts breakpoints which expect % of float.
# Do NOT fall back to finra_short here; leave as None when yfinance data is absent.
"short_percent": analyst.get("short_percent"),
"short_ratio": analyst.get("short_ratio"), "short_ratio": analyst.get("short_ratio"),
"piotroski_score": edgar.get("piotroski_score"), "piotroski_score": edgar.get("piotroski_score"),
"accruals_ratio": edgar.get("accruals_ratio"), "accruals_ratio": edgar.get("accruals_ratio"),

11
tests/conftest.py Normal file
View File

@ -0,0 +1,11 @@
"""
pytest configuration. Registers custom markers so -m "not network" works
without warnings.
"""
import pytest
def pytest_configure(config):
config.addinivalue_line(
"markers", "network: marks tests that require internet access (deselect with -m 'not network')"
)

99
tests/test_gui.py Normal file
View File

@ -0,0 +1,99 @@
"""
GUI smoke tests headless PySide6 (QT_QPA_PLATFORM=offscreen).
Verifies that all modules import and core widgets instantiate without crashing.
No network calls, no DB, no real data.
Run automatically via PostToolUse hook on every edit to screener_gui.py.
"""
import os
import sys
import subprocess
from pathlib import Path
PROJECT_ROOT = Path(__file__).parent.parent
DIST_DIR = PROJECT_ROOT / "dist"
PYTHON = sys.executable
def _run(code: str) -> subprocess.CompletedProcess:
"""Run Python code in a subprocess with offscreen Qt platform."""
env = os.environ.copy()
env["QT_QPA_PLATFORM"] = "offscreen"
return subprocess.run(
[PYTHON, "-c", code],
capture_output=True,
text=True,
cwd=str(DIST_DIR),
env=env,
timeout=30,
)
def test_stock_screener_imports():
"""stock_screener.py imports and exposes expected symbols."""
result = _run("""
import stock_screener as ss
required = [
'compute_value_score', 'compute_quality_score',
'compute_growth_score', 'compute_profitability_score',
'compute_sentiment_score', 'score_stocks',
'STRATEGY_PRESETS', 'WEIGHTS',
]
missing = [s for s in required if not hasattr(ss, s)]
assert not missing, f"Missing symbols: {missing}"
print("SCREENER_OK")
""")
assert "SCREENER_OK" in result.stdout, f"stock_screener import failed:\n{result.stderr}"
def test_screener_gui_imports():
"""screener_gui.py imports without error (PySide6, matplotlib, etc.)."""
result = _run("""
from PySide6.QtWidgets import QApplication
import sys
app = QApplication(sys.argv)
import screener_gui
print("GUI_IMPORT_OK")
""")
assert "GUI_IMPORT_OK" in result.stdout, (
f"screener_gui import failed:\n{result.stderr[-2000:]}"
)
def test_stock_table_model():
"""StockTableModel can be created and populated via set_data."""
result = _run("""
import sys
from PySide6.QtWidgets import QApplication
app = QApplication(sys.argv)
from screener_gui import StockTableModel
model = StockTableModel()
assert model.rowCount() == 0
model.set_data([("AAPL", "Apple Inc", 82.5, "Technology")])
assert model.rowCount() == 1
assert model.columnCount() == 5
print("TABLE_MODEL_OK")
""")
assert "TABLE_MODEL_OK" in result.stdout, (
f"StockTableModel test failed:\n{result.stderr[-2000:]}"
)
def test_screener_app_instantiates():
"""ScreenerApp (main window) instantiates without crashing."""
result = _run("""
import sys
from PySide6.QtWidgets import QApplication
app = QApplication(sys.argv)
from screener_gui import ScreenerApp
win = ScreenerApp()
# Verify expected attributes exist
assert hasattr(win, '_screener_page')
assert hasattr(win, '_search_page')
assert hasattr(win, '_sidebar')
print("APP_OK")
""")
assert "APP_OK" in result.stdout, (
f"ScreenerApp instantiation failed:\n{result.stderr[-2000:]}"
)

150
tests/test_network.py Normal file
View File

@ -0,0 +1,150 @@
"""
Network integration tests requires internet access and live API.
NOT run automatically on every edit (slow, rate-limited).
Run manually before pushing an update:
python -m pytest tests/test_network.py -v
Or run with the fast tests excluded:
python -m pytest tests/ -v -m "not network"
"""
import sys
from pathlib import Path
import pytest
import requests
sys.path.insert(0, str(Path(__file__).parent.parent / "dist"))
API_BASE = "https://api.verimundsolutions.com"
EDGAR_BASE = "https://data.sec.gov"
EDGAR_UA = {"User-Agent": "Verimund Solutions support@verimundsolutions.com"}
pytestmark = pytest.mark.network
# ---------------------------------------------------------------------------
# API health checks
# ---------------------------------------------------------------------------
class TestAPIHealth:
def test_update_endpoint_alive(self):
"""GET /update should respond (not timeout, not 500)."""
r = requests.get(f"{API_BASE}/update", timeout=10)
assert r.status_code != 500, f"/update returned 500: {r.text}"
assert r.status_code in (200, 400, 401, 422), (
f"Unexpected status from /update: {r.status_code}"
)
def test_versions_endpoint_alive(self):
"""GET /admin/versions returns 401 without auth (not 500 or timeout)."""
r = requests.get(f"{API_BASE}/admin/versions", timeout=10)
assert r.status_code in (200, 401, 403), (
f"/admin/versions returned unexpected status: {r.status_code}"
)
def test_auth_endpoint_rejects_bad_request(self):
"""POST /auth with garbage data returns 400/422, not 500."""
r = requests.post(
f"{API_BASE}/auth",
json={"license_key": "bad", "hwid": "bad", "timestamp": "0", "signature": "bad"},
timeout=10,
)
assert r.status_code in (400, 401, 422), (
f"/auth returned unexpected status: {r.status_code}"
)
def test_no_500_errors_on_any_public_endpoint(self):
endpoints = ["/update", "/admin/versions"]
for ep in endpoints:
r = requests.get(f"{API_BASE}{ep}", timeout=10)
assert r.status_code != 500, f"{ep} returned 500: {r.text}"
# ---------------------------------------------------------------------------
# yfinance field checks
# ---------------------------------------------------------------------------
class TestYFinanceFields:
"""
Verify that yfinance still returns the fields our scoring depends on.
Uses AAPL as a known-stable ticker.
If these fail, yfinance has renamed a field we use scoring will silently
return neutral 50 for affected metrics.
"""
@pytest.fixture(scope="class")
def aapl_info(self):
import yfinance as yf
return yf.Ticker("AAPL").info
REQUIRED_FIELDS = [
"trailingPE",
"forwardPE",
"priceToBook",
"returnOnAssets",
"returnOnEquity",
"currentRatio",
"debtToEquity",
"revenueGrowth",
"earningsGrowth",
"shortPercentOfFloat",
"recommendationMean",
"targetMeanPrice",
"currentPrice",
"marketCap",
]
def test_required_fields_present(self, aapl_info):
missing = [f for f in self.REQUIRED_FIELDS if f not in aapl_info]
assert not missing, (
f"yfinance dropped fields we depend on: {missing}\n"
f"Update scoring functions or field mappings."
)
def test_de_is_percentage_scale(self, aapl_info):
"""
yfinance returns debtToEquity as percentage (e.g. 150 not 1.5).
Our code divides by 100. If this breaks, D/E scoring silently becomes wrong.
"""
de = aapl_info.get("debtToEquity")
if de is not None:
assert de > 1.0, (
f"debtToEquity={de} looks like a ratio, not percentage. "
f"Remove the /100.0 normalization in stock_screener.py line ~1356."
)
def test_price_history_available(self):
import yfinance as yf
hist = yf.download("AAPL", period="1mo", progress=False, auto_adjust=True)
assert len(hist) > 10, "yfinance price history returned fewer than 10 bars for AAPL"
# ---------------------------------------------------------------------------
# EDGAR API checks
# ---------------------------------------------------------------------------
class TestEDGARAPI:
def test_frames_endpoint_alive(self):
"""EDGAR XBRL frames API returns valid data for a known concept/period."""
r = requests.get(
f"{EDGAR_BASE}/api/xbrl/frames/us-gaap/Revenues/USD/CY2023Q4.json",
headers=EDGAR_UA,
timeout=20,
)
assert r.status_code == 200, f"EDGAR frames API returned {r.status_code}"
data = r.json()
assert "data" in data, "EDGAR response missing 'data' key"
assert len(data["data"]) > 100, "EDGAR returned suspiciously few companies"
def test_company_facts_endpoint_alive(self):
"""EDGAR company facts for Apple (CIK 320193) returns valid data."""
r = requests.get(
f"{EDGAR_BASE}/api/xbrl/companyfacts/CIK0000320193.json",
headers=EDGAR_UA,
timeout=20,
)
assert r.status_code == 200, f"EDGAR company facts returned {r.status_code}"
data = r.json()
assert "facts" in data
assert "us-gaap" in data["facts"]

296
tests/test_scoring.py Normal file
View File

@ -0,0 +1,296 @@
"""
Fast scoring unit tests no DB, no network, no GUI.
Run automatically via PostToolUse hook on every edit to stock_screener.py.
Tests cover:
- Value score direction (cheap stocks score higher than expensive)
- Negative equity low quality score (not high)
- Piotroski score feeds into quality correctly
- Sentiment confidence weighting (1 headline 10 headlines)
- All-None row doesn't crash any scoring function
- All scores are within 0100 bounds
- Profile classification (early_stage, financial, high_growth, mature)
"""
import sys
from pathlib import Path
import pandas as pd
import numpy as np
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent / "dist"))
from stock_screener import (
compute_value_score,
compute_growth_score,
compute_quality_score,
compute_profitability_score,
compute_sentiment_score,
compute_analyst_score,
)
# ---------------------------------------------------------------------------
# Mock sector stats — median/MAD pairs per metric for "Technology" sector.
# Used so z-score functions return meaningful values instead of neutral 50.
# ---------------------------------------------------------------------------
MOCK_SS = {
"Technology": {
"pe_trailing": (25.0, 15.0),
"pe_forward": (22.0, 12.0),
"pb_ratio": (5.0, 3.0),
"ev_ebitda": (18.0, 10.0),
"ev_revenue": (8.0, 5.0),
"revenue_growth": (0.12, 0.10),
"earnings_growth": (0.15, 0.12),
"eps_growth": (0.10, 0.15),
"roa": (0.08, 0.06),
"roe": (0.15, 0.10),
"debt_to_equity": (0.50, 0.40),
"current_ratio": (2.0, 0.8),
"fcf_yield": (0.04, 0.03),
"operating_margin": (0.15, 0.10),
"profit_margin": (0.12, 0.08),
"analyst_upside": (0.10, 0.15),
}
}
def _base_row(**overrides):
"""Return a dict with all required fields set to safe defaults."""
defaults = dict(
sector="Technology",
pe_trailing=25.0, pe_forward=22.0,
pb_ratio=5.0, ev_ebitda=18.0, ev_revenue=8.0,
revenue_growth=0.12, earnings_growth=0.15,
eps_trailing=3.0, eps_forward=3.5,
roa=0.08, roe=0.15,
debt_to_equity=0.50, current_ratio=2.0,
piotroski_score=2,
fcf_yield=0.04, operating_margin=0.15,
profit_margin=0.12, accruals_ratio=-0.02,
news_sentiment=0.0, news_headline_count=5,
analyst_count=8, analyst_upside=0.10,
analyst_norm=50.0,
revenue=500_000_000, market_cap=5_000_000_000,
short_percent=0.03, volatility=0.25, beta=1.0,
)
defaults.update(overrides)
return defaults
def _df(*rows):
"""Build a DataFrame from _base_row dicts, indexed 0..n."""
return pd.DataFrame(list(rows)).reset_index(drop=True)
# ---------------------------------------------------------------------------
# Value score tests
# ---------------------------------------------------------------------------
class TestValueScore:
def test_cheap_beats_expensive(self):
df = _df(
_base_row(pe_trailing=8, pe_forward=7, pb_ratio=1.5, ev_ebitda=6), # cheap
_base_row(pe_trailing=120, pe_forward=100, pb_ratio=20, ev_ebitda=60), # expensive
)
scores = compute_value_score(df, MOCK_SS)
assert scores.iloc[0] > scores.iloc[1], (
f"Cheap stock ({scores.iloc[0]:.1f}) should beat expensive ({scores.iloc[1]:.1f})"
)
def test_none_pe_doesnt_crash(self):
df = _df(_base_row(pe_trailing=None, pe_forward=None))
scores = compute_value_score(df, MOCK_SS)
assert 0 <= float(scores.iloc[0]) <= 100
def test_scores_in_bounds(self):
rows = [
_base_row(pe_trailing=5),
_base_row(pe_trailing=200),
_base_row(pe_trailing=None),
]
scores = compute_value_score(_df(*rows), MOCK_SS)
assert scores.between(0, 100).all(), f"Out-of-bounds scores: {scores.tolist()}"
# ---------------------------------------------------------------------------
# Quality score tests
# ---------------------------------------------------------------------------
class TestQualityScore:
def test_negative_equity_scores_low(self):
"""Negative D/E (negative book equity) must score low, not high."""
df = _df(
_base_row(debt_to_equity=-2.0, roa=0.05), # negative equity
_base_row(debt_to_equity=0.3, roa=0.12), # healthy balance sheet
)
scores = compute_quality_score(df, MOCK_SS)
assert scores.iloc[0] < 40, (
f"Negative equity should score < 40, got {scores.iloc[0]:.1f}"
)
assert scores.iloc[1] > scores.iloc[0], (
f"Healthy balance sheet ({scores.iloc[1]:.1f}) should beat negative equity ({scores.iloc[0]:.1f})"
)
def test_high_piotroski_beats_low(self):
df = _df(
_base_row(piotroski_score=4), # strong
_base_row(piotroski_score=0), # weak
)
scores = compute_quality_score(df, MOCK_SS)
assert scores.iloc[0] > scores.iloc[1], (
f"Piotroski 4 ({scores.iloc[0]:.1f}) should beat Piotroski 0 ({scores.iloc[1]:.1f})"
)
def test_high_de_scores_lower(self):
df = _df(
_base_row(debt_to_equity=0.1), # low leverage
_base_row(debt_to_equity=5.0), # high leverage
)
scores = compute_quality_score(df, MOCK_SS)
assert scores.iloc[0] > scores.iloc[1], (
f"Low D/E ({scores.iloc[0]:.1f}) should beat high D/E ({scores.iloc[1]:.1f})"
)
def test_scores_in_bounds(self):
rows = [
_base_row(debt_to_equity=-5.0),
_base_row(debt_to_equity=999.0),
_base_row(roa=None, roe=None),
]
scores = compute_quality_score(_df(*rows), MOCK_SS)
assert scores.between(0, 100).all(), f"Out-of-bounds scores: {scores.tolist()}"
# ---------------------------------------------------------------------------
# Sentiment score tests
# ---------------------------------------------------------------------------
class TestSentimentScore:
def test_confidence_weighting(self):
"""1 headline should blend toward neutral vs 10 headlines."""
df = _df(
_base_row(news_sentiment=0.8, news_headline_count=1), # low confidence
_base_row(news_sentiment=0.8, news_headline_count=10), # high confidence
)
scores = compute_sentiment_score(df)
assert scores.iloc[1] > scores.iloc[0], (
f"10 headlines ({scores.iloc[1]:.1f}) should outscore 1 headline ({scores.iloc[0]:.1f}) "
f"for same positive sentiment"
)
def test_single_headline_blends_toward_neutral(self):
df = _df(_base_row(news_sentiment=0.8, news_headline_count=1))
score = float(compute_sentiment_score(df).iloc[0])
# 1 headline → confidence=0.2, so score should be closer to 50 than to 80+
assert score < 70, f"1-headline score should blend toward neutral, got {score:.1f}"
def test_negative_sentiment_scores_low(self):
df = _df(
_base_row(news_sentiment=-0.7, news_headline_count=5),
_base_row(news_sentiment=0.7, news_headline_count=5),
)
scores = compute_sentiment_score(df)
assert scores.iloc[0] < scores.iloc[1]
def test_scores_in_bounds(self):
rows = [
_base_row(news_sentiment=-1.0, news_headline_count=0),
_base_row(news_sentiment=1.0, news_headline_count=20),
_base_row(news_sentiment=None, news_headline_count=None),
]
scores = compute_sentiment_score(_df(*rows))
assert scores.between(0, 100).all()
# ---------------------------------------------------------------------------
# Growth score tests
# ---------------------------------------------------------------------------
class TestGrowthScore:
def test_high_growth_beats_low(self):
df = _df(
_base_row(revenue_growth=0.50, earnings_growth=0.60),
_base_row(revenue_growth=-0.10, earnings_growth=-0.20),
)
scores = compute_growth_score(df, MOCK_SS)
assert scores.iloc[0] > scores.iloc[1]
def test_scores_in_bounds(self):
rows = [
_base_row(revenue_growth=None, earnings_growth=None),
_base_row(revenue_growth=5.0, earnings_growth=5.0),
_base_row(revenue_growth=-1.0, earnings_growth=-1.0),
]
scores = compute_growth_score(_df(*rows), MOCK_SS)
assert scores.between(0, 100).all()
# ---------------------------------------------------------------------------
# Profitability score tests
# ---------------------------------------------------------------------------
class TestProfitabilityScore:
def test_cash_backed_earnings_score_higher(self):
"""Negative accruals (cash-backed earnings) should beat high accruals."""
df = _df(
_base_row(accruals_ratio=-0.20, operating_margin=0.20), # cash-backed
_base_row(accruals_ratio=0.20, operating_margin=0.20), # accrual-heavy
)
scores = compute_profitability_score(df, MOCK_SS)
assert scores.iloc[0] > scores.iloc[1]
def test_scores_in_bounds(self):
rows = [
_base_row(fcf_yield=None, operating_margin=None, profit_margin=None),
_base_row(accruals_ratio=-0.5),
_base_row(accruals_ratio=0.5),
]
scores = compute_profitability_score(_df(*rows), MOCK_SS)
assert scores.between(0, 100).all()
# ---------------------------------------------------------------------------
# All-None row doesn't crash any scorer
# ---------------------------------------------------------------------------
class TestNoneRobustness:
NULL_ROW = dict(
sector=None, pe_trailing=None, pe_forward=None,
pb_ratio=None, ev_ebitda=None, ev_revenue=None,
revenue_growth=None, earnings_growth=None,
eps_trailing=None, eps_forward=None,
roa=None, roe=None, debt_to_equity=None,
current_ratio=None, piotroski_score=None,
fcf_yield=None, operating_margin=None,
profit_margin=None, accruals_ratio=None,
news_sentiment=None, news_headline_count=None,
analyst_count=None, analyst_upside=None, analyst_norm=None,
revenue=None, market_cap=None,
short_percent=None, volatility=None, beta=None,
)
def test_value_no_crash(self):
df = pd.DataFrame([self.NULL_ROW])
scores = compute_value_score(df, {})
assert scores.between(0, 100).all()
def test_quality_no_crash(self):
df = pd.DataFrame([self.NULL_ROW])
scores = compute_quality_score(df, {})
assert scores.between(0, 100).all()
def test_growth_no_crash(self):
df = pd.DataFrame([self.NULL_ROW])
scores = compute_growth_score(df, {})
assert scores.between(0, 100).all()
def test_profitability_no_crash(self):
df = pd.DataFrame([self.NULL_ROW])
scores = compute_profitability_score(df, {})
assert scores.between(0, 100).all()
def test_sentiment_no_crash(self):
df = pd.DataFrame([self.NULL_ROW])
scores = compute_sentiment_score(df)
assert scores.between(0, 100).all()

1245
ui_mockup.html Normal file

File diff suppressed because it is too large Load Diff

View File

@ -19,7 +19,7 @@ import requests
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Current app version — kept in sync by push_update.py before each build. # Current app version — kept in sync by push_update.py before each build.
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
APP_VERSION = "1.4.0" APP_VERSION = "1.4.1"
def _exe_dir() -> str: def _exe_dir() -> str: