Magic Formula screener: earnings yield, capital returns and ranking

from datetime import datetime
from decimal import Decimal, InvalidOperation
from fractions import Fraction


def timestamp(value):
    if not isinstance(value, str):
        raise ValueError("timestamp must be a string")
    result = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if result.utcoffset() is None:
        raise ValueError("timestamp requires a UTC offset")
    return result


def number(value):
    if not isinstance(value, str):
        raise ValueError("numeric inputs must be decimal strings")
    result = Decimal(value)
    if not result.is_finite():
        raise ValueError("non-finite input")
    return Fraction(result)


def screen(rows, cutoff):
    cutoff = timestamp(cutoff)
    valid, excluded, seen = [], [], set()
    for row in rows:
        symbol = row["symbol"]
        if symbol in seen:
            raise ValueError("duplicate issuer identifier: " + symbol)
        seen.add(symbol)
        try:
            if row.get("reviewed") is not True:
                raise ValueError("eligibility/accounting review incomplete")
            if row["sector"] in {"Financial Services", "Utilities"}:
                raise ValueError("excluded sector")
            if timestamp(row["available_at"]) > cutoff:
                raise ValueError("inputs unavailable at cutoff")
            ebit, ev, nwc, nfa = [number(row[k])
                                  for k in ("ebit", "ev", "nwc", "nfa")]
            capital = nwc + nfa
            if ebit <= 0 or ev <= 0 or capital <= 0 or nfa < 0:
                raise ValueError("invalid EBIT, EV or tangible capital")
            valid.append(dict(symbol=symbol, ey=ebit / ev,
                              roc=ebit / capital))
        except (KeyError, TypeError, ValueError, InvalidOperation) as error:
            excluded.append((symbol, str(error)))
    for field in ("ey", "roc"):
        for row in valid:
            better = sum(other[field] > row[field] for other in valid)
            equal = sum(other[field] == row[field] for other in valid)
            row[field + "_rank"] = better + Fraction(equal + 1, 2)
    for row in valid:
        row["score"] = row["ey_rank"] + row["roc_rank"]
    return sorted(valid, key=lambda r: (r["score"], r["symbol"])), excluded


def shortlist(ranked, count):
    if not isinstance(count, int) or isinstance(count, bool) or count < 1:
        raise ValueError("count must be a positive integer")
    if not ranked:
        return []
    boundary = ranked[min(count, len(ranked)) - 1]["score"]
    return [row for row in ranked if row["score"] <= boundary]


inputs = [
    ("A", "60", "600", "70", "130"),
    ("B", "80", "1000", "80", "120"),
    ("C", "40", "500", "40", "60"),
    ("D", "75", "500", "200", "300"),
    ("E", "30", "500", "-50", "200"),
]
rows = [dict(symbol=s, ebit=e, ev=v, nwc=w, nfa=f,
             sector="Industrials", reviewed=True,
             available_at="2026-10-16T20:00:00Z")
        for s, e, v, w, f in inputs]
ranked, excluded = screen(rows, "2026-10-16T20:00:00Z")
print([(r["symbol"], str(r["score"])) for r in ranked])
# [('A', '5'), ('B', '5'), ('C', '5'), ('D', '6'), ('E', '9')]
print([r["symbol"] for r in shortlist(ranked, 2)])
# ['A', 'B', 'C']
print(excluded)  # []