"""手动模拟定时选股；调度默认关闭，不注册任何系统任务。"""
import copy
import json
from datetime import datetime
from decimal import Decimal as D, InvalidOperation

SCHEDULER_ENABLED = False
CALENDAR = {"2026-10-19", "2026-10-20", "2026-10-21"}


class Screen:
    def __init__(self):
        self.done = set()
        self.last_success = {"date": "2026-10-19", "symbols": ["AAA"]}
        self.events = []

    def refresh(self, day, payload):
        self.events.append("refresh:"+day)
        if isinstance(payload, Exception):
            raise payload
        return copy.deepcopy(payload)

    def validate(self, rows, day):
        self.events.append("validate")
        if set(rows) != {"AAA", "BBB"}:
            raise ValueError("incomplete_universe")
        for row in rows.values():
            if row.get("date") != day:
                raise ValueError("stale_or_missing_date")
            if row.get("pe") is None:
                raise ValueError("missing_pe")
            try:
                pe = D(str(row["pe"]))
            except (InvalidOperation, ValueError, TypeError) as exc:
                raise ValueError("invalid_pe") from exc
            if not pe.is_finite() or pe <= 0:
                raise ValueError("invalid_pe")

    def screen(self, rows):
        self.events.append("screen")
        return sorted(symbol for symbol, row in rows.items() if D(str(row["pe"])) < 20)

    def run(self, trigger, payload, dry_run=True):
        start = len(self.events)
        now = datetime.fromisoformat(trigger)
        if now.utcoffset() is None or now.utcoffset().total_seconds() != 8*3600:
            raise ValueError("expected_fixed_plus08_timezone")
        day = str(now.date())
        slot = "screen-v1:"+day
        result = {"slot": slot, "current_result": None, "last_success": copy.deepcopy(self.last_success)}
        if not dry_run and not SCHEDULER_ENABLED:
            return {**result, "status": "scheduler_disabled", "trace": []}
        if day not in CALENDAR:
            return {**result, "status": "closed", "trace": []}
        if (now.hour, now.minute) < (16, 0):
            return {**result, "status": "not_due", "trace": []}
        if slot in self.done:
            return {**result, "status": "duplicate_suppressed", "trace": []}
        try:
            rows = self.refresh(day, payload)
            self.validate(rows, day)
            symbols = self.screen(rows)
            self.events.append("diff")
            previous = set(self.last_success["symbols"])
            current = {"date": day, "symbols": symbols, "added": sorted(set(symbols)-previous),
                       "removed": sorted(previous-set(symbols))}
            self.events.append("publish_in_memory")
            self.last_success = copy.deepcopy(current)
            self.done.add(slot)
            result.update(status="dry_run_success", current_result=current)
        except (ValueError, TimeoutError) as exc:
            result.update(status="blocked", reason=str(exc))
        result["trace"] = self.events[start:]
        return result


data = {"AAA": {"date": "2026-10-20", "pe": "25"}, "BBB": {"date": "2026-10-20", "pe": "18"}}
job = Screen()
main = job.run("2026-10-20T16:30:00+08:00", data)
assert main["current_result"]["symbols"] == ["BBB"]
assert main["current_result"]["added"] == ["BBB"] and main["current_result"]["removed"] == ["AAA"]
assert main["trace"] == ["refresh:2026-10-20", "validate", "screen", "diff", "publish_in_memory"]
duplicate = job.run("2026-10-20T17:00:00+08:00", data)
assert duplicate["status"] == "duplicate_suppressed" and duplicate["trace"] == []
closed = job.run("2026-10-24T16:30:00+08:00", data)
assert closed["status"] == "closed" and closed["trace"] == []
failures = []
for bad in [TimeoutError("provider_timeout"), {**data, "AAA": {"date": "2026-10-19", "pe": "18"}},
            {**data, "AAA": {"date": "2026-10-20"}}]:
    fresh = Screen()
    blocked = fresh.run("2026-10-20T16:30:00+08:00", bad)
    assert blocked["status"] == "blocked" and blocked["current_result"] is None
    assert fresh.last_success["date"] == "2026-10-19" and not fresh.done
    failures.append(blocked)
for token in ["NaN", "Infinity", "-Infinity", "0", "-1"]:
    assert Screen().run("2026-10-20T16:30:00+08:00", {**data, "AAA": {"date": "2026-10-20", "pe": token}})["status"] == "blocked"
assert Screen().run("2026-10-20T16:30:00+08:00", data, dry_run=False)["status"] == "scheduler_disabled"
print(json.dumps({"main": main, "duplicate": duplicate, "closed": closed, "failures": failures,
                  "scheduler_enabled": SCHEDULER_ENABLED}, sort_keys=True))