ClaudeandClaude Opus 5.5
4972fa40a6
New report type annual_full on the personal report queue, like
synastry_full. POST /api/reports {reportType: annual_full} reserves
annual.full (20 credits by default, admin-editable), stores the range
and the optional situation, and queues the job; a year already written
for the same subject opens directly without a second charge. The worker
asks /api/annual/report_facts for the server cards, writes the sections
one at a time (person, overview, domains, windows, remaining, close)
with a program gate and one rewrite, saves the snapshot after every
section, then settles; a failure releases the reservation.
GET /api/annual/overview serves the overlay from the slim package.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DbBXoyeZrYn6ryNcSe5Y2Y
656 lines
34 KiB
Python
656 lines
34 KiB
Python
"""Annual report facts and claim cards (Gitea #153, on top of annual package v1).
|
|
|
|
The annual report's server-side rule layer. It reads the annual package and
|
|
the natal chart and returns the cards the report writer may use: who the
|
|
person is (natal base), the year's phases with a lean, turning points,
|
|
candidate windows, per-domain signals, conflicts and the technique audit.
|
|
The model only words these cards; it cannot add a conclusion or raise one.
|
|
|
|
Every default the #152 ruling left open lives in ``CONFIG`` and
|
|
``HOUSE_DOMAINS`` (one place each) and is listed in
|
|
``docs/engine/annual-report-v1.md`` as pending astrologer confirmation:
|
|
|
|
1. Narayana skeleton = the legacy profile the product timeline uses; the
|
|
Rath profile is shown as reference and never counts as support.
|
|
2. Tithi Pravesha stays an independent view; it never enters windows.
|
|
3. Contacts are exact conjunctions, ingresses and stations only (no orbs).
|
|
4. Lean = functional roles (natal Lagna) of the natal skeleton lords
|
|
(Vimshottari AD and PD, Narayana AD sign lord), one count per planet.
|
|
Annual-side lords (Mudda, Patyayini) are listed as the year's unfolding
|
|
and never vote (the ruling: 两边各负其责，不做简单投票).
|
|
5. Windows: computed boundaries and contacts within 21 days of each other,
|
|
at least two independent systems; Rahu/Ketu are one axis, Mudda and
|
|
Patyayini one annual layer, all Vimshottari levels one system.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from typing import Any
|
|
|
|
try:
|
|
import annual_package as ap
|
|
except ImportError: # pragma: no cover
|
|
from scripts import annual_package as ap
|
|
|
|
SCHEMA = "jyotish.annual_report_facts.v1"
|
|
|
|
CONFIG: dict[str, Any] = {
|
|
"narayana_skeleton_profile": "legacy",
|
|
"narayana_rath_counts_as_support": False,
|
|
"tithi_pravesha_in_windows": False,
|
|
"contact_rule": "exact_conjunction_ingress_station_only",
|
|
"lean_rule": "functional_roles_of_natal_skeleton_lords_v1",
|
|
"lean_lords": ("vimshottari_ad", "vimshottari_pd", "narayana_ad"),
|
|
"phase_min_days": 21,
|
|
"window_cluster_days": 21,
|
|
"window_max_span_days": 45,
|
|
"window_min_independent_systems": 2,
|
|
"window_target": 6,
|
|
"window_hard_cap": 8,
|
|
"unconfirmed_birth_widen_days": 14,
|
|
"lookback_months": ap.LOOKBACK_MONTHS,
|
|
}
|
|
|
|
# House -> report domain. One house may feed two domains.
|
|
HOUSE_DOMAINS: dict[str, tuple[int, ...]] = {
|
|
"career": (10, 6),
|
|
"wealth": (2, 11),
|
|
"relationship": (7, 5),
|
|
"home": (4,),
|
|
"energy": (1, 6, 12),
|
|
"learning": (4, 5, 9),
|
|
}
|
|
DOMAIN_VARGAS: dict[str, tuple[int, ...]] = {
|
|
"career": (10,),
|
|
"wealth": (2, 11),
|
|
"relationship": (9,),
|
|
"home": (4,),
|
|
"energy": (),
|
|
"learning": (24,),
|
|
}
|
|
# Divisions whose sign changes within minutes; flagged when the birth time is not confirmed.
|
|
TIME_SENSITIVE_DIVISIONS = frozenset({10, 11, 24})
|
|
GRAHAS = ("Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn", "Rahu", "Ketu")
|
|
SYSTEM_OF_BODY = {"Jupiter": "transit_jupiter", "Saturn": "transit_saturn", "Rahu": "transit_nodes", "Ketu": "transit_nodes"}
|
|
# Systems whose boundaries move with the birth clock (Moon / Lagna based).
|
|
BIRTH_TIME_SENSITIVE_SYSTEMS = frozenset({"vimshottari", "narayana"})
|
|
YOGA_CATEGORIES_FIRST = ("pancha_mahapurusha", "raja", "dhana", "viparita")
|
|
MAX_YOGAS = 8
|
|
|
|
|
|
def _jd(boundary: Any) -> float | None:
|
|
if isinstance(boundary, dict) and isinstance(boundary.get("jd_ut"), (int, float)):
|
|
return float(boundary["jd_ut"])
|
|
return None
|
|
|
|
|
|
def _owned(lagna_idx: int, planet: str) -> list[int]:
|
|
return [house for house in range(1, 13)
|
|
if ap.SIGN_LORDS[ap.SIGNS[(lagna_idx + house - 1) % 12]] == planet]
|
|
|
|
|
|
def _domains_for(houses: list[int]) -> list[str]:
|
|
return [domain for domain, owned in HOUSE_DOMAINS.items() if any(h in owned for h in houses)]
|
|
|
|
|
|
# ── natal base ─────────────────────────────────────────────────────────────
|
|
|
|
def _natal_base(ctx: Any, birth_time_confirmed: bool) -> dict[str, Any]:
|
|
engine = ap._engine()
|
|
chart, asc_idx, _jd_value, _ay = engine._compute_chart_from_args(ctx.engine_args())
|
|
planets = chart.get("planets") or {}
|
|
asc = chart.get("ascendant") or {}
|
|
lagna_idx = int(asc_idx)
|
|
lons = {name: float(row["degree_raw"]) for name, row in planets.items()
|
|
if isinstance(row, dict) and row.get("degree_raw") is not None}
|
|
houses_of = {name: int(row["house"]) for name, row in planets.items()
|
|
if isinstance(row, dict) and isinstance(row.get("house"), int)}
|
|
functional = _module_call("functional_layer", "functional_layer_for_ascendant", asc)
|
|
house_rows = []
|
|
for house in range(1, 13):
|
|
sign = ap.SIGNS[(lagna_idx + house - 1) % 12]
|
|
lord = ap.SIGN_LORDS[sign]
|
|
house_rows.append({
|
|
"house": house, "sign": sign, "lord": lord,
|
|
"lord_house": houses_of.get(lord), "lord_sign": (planets.get(lord) or {}).get("sign"),
|
|
"occupants": [name for name in GRAHAS if houses_of.get(name) == house],
|
|
})
|
|
domains = {}
|
|
for domain, houses in HOUSE_DOMAINS.items():
|
|
vargas = [_varga_cue(division, float(asc.get("degree_raw")), lons, birth_time_confirmed)
|
|
for division in DOMAIN_VARGAS[domain]]
|
|
domains[domain] = {
|
|
"houses": [row for row in house_rows if row["house"] in houses],
|
|
"vargas": vargas,
|
|
"special": _special_points(domain, lagna_idx, lons),
|
|
}
|
|
return {
|
|
"status": "computed",
|
|
"producer": "jyotish_engine._compute_chart_from_args",
|
|
"lagna": {"sign": ap.SIGNS[lagna_idx], "degree_in_sign": float(asc.get("degree_raw")) % 30.0,
|
|
"lord": ap.SIGN_LORDS[ap.SIGNS[lagna_idx]],
|
|
"lord_house": houses_of.get(ap.SIGN_LORDS[ap.SIGNS[lagna_idx]])},
|
|
"moon": {"sign": (planets.get("Moon") or {}).get("sign"), "house": houses_of.get("Moon"),
|
|
"nakshatra": (planets.get("Moon") or {}).get("nakshatra")},
|
|
"planets": {name: {"sign": (planets.get(name) or {}).get("sign"), "house": houses_of.get(name),
|
|
"retrograde": (planets.get(name) or {}).get("retrograde"),
|
|
"dignity": (planets.get(name) or {}).get("status")}
|
|
for name in GRAHAS if name in planets},
|
|
"functional": _functional_block(functional),
|
|
"houses": house_rows,
|
|
"yogas": _yogas(planets, ap.SIGNS[lagna_idx]),
|
|
"domains": domains,
|
|
"birth_time_confirmed": bool(birth_time_confirmed),
|
|
"_lagna_idx": lagna_idx,
|
|
"_houses_of": houses_of,
|
|
}
|
|
|
|
|
|
def _module_call(module: str, name: str, *args: Any) -> Any:
|
|
return getattr(ap._module(module), name)(*args)
|
|
|
|
|
|
def _functional_block(raw: Any) -> dict[str, Any]:
|
|
if not isinstance(raw, dict) or raw.get("status") != "used":
|
|
return ap.block("blocked", "functional_layer.functional_layer_for_ascendant",
|
|
reason=str((raw or {}).get("effect_on_confidence") or "functional_layer_unavailable"))
|
|
return ap.block(
|
|
"computed", "functional_layer.functional_layer_for_ascendant",
|
|
benefics=raw.get("functional_benefics") or [], malefics=raw.get("functional_malefics") or [],
|
|
neutrals=raw.get("functional_neutrals") or [], yogakarakas=raw.get("yogakarakas") or [],
|
|
lagna_lord=raw.get("lagna_lord"), owned_houses=raw.get("owned_houses") or {},
|
|
profile=raw.get("profile"), nodes_note="罗计不单独定功能吉凶",
|
|
)
|
|
|
|
|
|
def _varga_cue(division: int, asc_lon: float, lons: dict[str, float], confirmed: bool) -> dict[str, Any]:
|
|
varga = ap._module("varga")
|
|
lagna = varga.calc_varga(asc_lon, division)
|
|
lord = lagna.get("lord")
|
|
lord_lon = lons.get(lord)
|
|
lord_varga = varga.calc_varga(lord_lon, division) if isinstance(lord_lon, (int, float)) else {}
|
|
lord_house = ((int(lord_varga["sign_idx"]) - int(lagna["sign_idx"])) % 12 + 1) if lord_varga else None
|
|
sensitive = division in TIME_SENSITIVE_DIVISIONS and not confirmed
|
|
return {
|
|
"division": f"D{division}",
|
|
"status": "blocked" if sensitive else "computed",
|
|
"producer": "varga.calc_varga",
|
|
**({"reason": "birth_time_not_confirmed"} if sensitive else {}),
|
|
"lagna_sign": lagna.get("sign"), "lagna_lord": lord,
|
|
"lord_sign": lord_varga.get("sign"), "lord_house": lord_house,
|
|
}
|
|
|
|
|
|
def _special_points(domain: str, lagna_idx: int, lons: dict[str, float]) -> dict[str, Any]:
|
|
jaimini = ap._module("jaimini")
|
|
padas = jaimini.calc_arudha_padas(lagna_idx, lons).get("padas") or {}
|
|
out: dict[str, Any] = {}
|
|
if domain == "career" and padas.get("A10"):
|
|
out["A10"] = padas["A10"].get("sign")
|
|
if domain == "relationship":
|
|
if padas.get("A7"):
|
|
out["A7"] = padas["A7"].get("sign")
|
|
if padas.get("UL"):
|
|
out["UL"] = padas["UL"].get("sign")
|
|
degrees = {name: lon % 30.0 for name, lon in lons.items() if name in GRAHAS}
|
|
dk = (jaimini.calc_chara_karaka_7(degrees).get("karaka_table") or {}).get("Darakaraka") or {}
|
|
if dk.get("planet"):
|
|
out["DK"] = dk["planet"]
|
|
return out
|
|
|
|
|
|
def _yogas(planets: dict, lagna_sign: str) -> dict[str, Any]:
|
|
try:
|
|
rows = ap._module("yoga_engine").detect_yogas(planets, lagna_sign) or []
|
|
except Exception as exc: # noqa: BLE001 - a yoga failure is a blocked block
|
|
return ap.block("blocked", "yoga_engine.detect_yogas", reason=f"{exc.__class__.__name__}: {exc}")
|
|
def rank(row: dict) -> tuple[int, int]:
|
|
category = str(row.get("category") or "")
|
|
first = YOGA_CATEGORIES_FIRST.index(category) if category in YOGA_CATEGORIES_FIRST else len(YOGA_CATEGORIES_FIRST)
|
|
strength = {"强": 0, "中": 1}.get(str(row.get("strength")), 2)
|
|
return (strength, first)
|
|
picked = sorted((row for row in rows if isinstance(row, dict) and row.get("name")), key=rank)[:MAX_YOGAS]
|
|
# Structure only: name, category and the combination. The producer's
|
|
# effect phrases are not facts the report may repeat.
|
|
return ap.block("computed", "yoga_engine.detect_yogas", total=len(rows), rows=[
|
|
{"name": row.get("name"), "name_cn": row.get("name_cn"), "category": row.get("category"),
|
|
"combination": row.get("combination"), "strength": row.get("strength")} for row in picked
|
|
])
|
|
|
|
|
|
def _role(functional: dict, planet: str | None) -> str:
|
|
if not planet or functional.get("status") != "computed" or planet in ("Rahu", "Ketu"):
|
|
return "none"
|
|
if planet in functional.get("yogakarakas", []) or planet in functional.get("benefics", []):
|
|
return "support"
|
|
if planet in functional.get("malefics", []):
|
|
return "pressure"
|
|
return "none"
|
|
|
|
|
|
# ── cards ──────────────────────────────────────────────────────────────────
|
|
|
|
def _rows(block: Any, key: str = "rows") -> list[dict]:
|
|
return [row for row in (block or {}).get(key) or [] if isinstance(row, dict)] if isinstance(block, dict) else []
|
|
|
|
|
|
def _active(rows: list[dict], jd: float) -> dict | None:
|
|
for row in rows:
|
|
start, end = _jd(row.get("start")), _jd(row.get("end"))
|
|
if start is not None and end is not None and start <= jd < end:
|
|
return row
|
|
return None
|
|
|
|
|
|
def _overlapping(rows: list[dict], lo: float, hi: float) -> list[dict]:
|
|
out = []
|
|
for row in rows:
|
|
start, end = _jd(row.get("start")), _jd(row.get("end"))
|
|
if start is not None and end is not None and start < hi and end > lo:
|
|
out.append(row)
|
|
return out
|
|
|
|
|
|
def _annual_house(chart_block: dict, planet: str | None) -> int | None:
|
|
if planet == "Lagna":
|
|
return 1
|
|
return ((chart_block.get("planets") or {}).get(planet) or {}).get("house") if planet else None
|
|
|
|
|
|
def build_cards(package: dict, natal: dict) -> dict[str, Any]:
|
|
zone_jd = _jd(package["report_range"].get("start"))
|
|
if package["report_range"].get("status") != "computed" or zone_jd is None:
|
|
return ap.block("blocked", "annual_report_facts.build_cards", reason="report_range_blocked")
|
|
range_lo = _jd(package["report_range"]["start"])
|
|
range_hi = _jd(package["report_range"]["end"])
|
|
obs_jd = _jd(package["report_range"]["observation"]["instant"])
|
|
lookback_jd = _jd(package["report_range"]["observation"]["lookback_start"])
|
|
lagna_idx = natal["_lagna_idx"]
|
|
houses_of = natal["_houses_of"]
|
|
functional = natal["functional"]
|
|
confirmed = natal["birth_time_confirmed"]
|
|
|
|
vim_rows = _rows(package["natal_periods"].get("vimshottari"))
|
|
narayana = package["natal_periods"].get("narayana") or {}
|
|
legacy_rows = _rows(narayana.get("legacy"))
|
|
rath_rows = _rows(narayana.get("rath"))
|
|
skeleton_rows = legacy_rows if CONFIG["narayana_skeleton_profile"] == "legacy" else rath_rows
|
|
charts = [chart for chart in package["year_charts"] if chart.get("status") == "computed"]
|
|
|
|
def natal_houses(planet: str | None) -> list[int]:
|
|
if not planet:
|
|
return []
|
|
return sorted(set(_owned(lagna_idx, planet) + ([houses_of[planet]] if planet in houses_of else [])))
|
|
|
|
def sign_house(sign: str | None) -> int | None:
|
|
return ((ap.SIGNS.index(sign) - lagna_idx) % 12 + 1) if sign in ap.SIGNS else None
|
|
|
|
def chart_at(jd: float) -> dict | None:
|
|
for chart in charts:
|
|
if _jd(chart["validity"]["start"]) <= jd < _jd(chart["validity"]["end"]):
|
|
return chart
|
|
return None
|
|
|
|
# Turning points (computed boundaries and exact contacts inside the range).
|
|
events: list[dict[str, Any]] = []
|
|
|
|
def add(jd: float | None, instant: Any, system: str, kind: str, houses: list[int], **detail: Any) -> None:
|
|
if jd is None or not (range_lo <= jd < range_hi):
|
|
return
|
|
events.append({"jd": jd, "instant": instant, "system": system, "kind": kind,
|
|
"houses": sorted(set(h for h in houses if h)), "domains": _domains_for(houses),
|
|
"birth_time_sensitive": (system in BIRTH_TIME_SENSITIVE_SYSTEMS or "lagna_house" in detail.get("basis", ""))
|
|
and not confirmed,
|
|
**detail})
|
|
|
|
for chart in charts:
|
|
blocks = chart["blocks"]
|
|
start = chart["validity"]["start"]
|
|
if _jd(start) > range_lo:
|
|
muntha = (blocks["muntha"].get("data") or {}) if blocks["muntha"]["status"] == "computed" else {}
|
|
add(_jd(start), start, "tajika_year", "year_chart_change", [muntha.get("muntha_house_from_annual_ascendant")],
|
|
target_year=chart["target_year"], basis="annual_chart")
|
|
for row in _rows(blocks["mudda_dasha"], "periods"):
|
|
lord = row.get("lord")
|
|
add(_jd(row["start"]), row["start"], "tajika_annual", "mudda_change",
|
|
[_annual_house(blocks["chart"], lord)], lord=lord, target_year=chart["target_year"], basis="annual_chart")
|
|
for row in _rows(blocks["patyayini"], "periods"):
|
|
lord = row.get("lord")
|
|
add(_jd(row["start"]), row["start"], "tajika_annual", "patyayini_change",
|
|
[_annual_house(blocks["chart"], lord)], lord=lord, target_year=chart["target_year"], basis="annual_chart")
|
|
for row in vim_rows:
|
|
if row.get("level") in ("AD", "PD"):
|
|
add(_jd(row["start"]), row["start"], "vimshottari", f"{row['level'].lower()}_change",
|
|
natal_houses(row.get("lord")), lord=row.get("lord"), level=row.get("level"), basis="natal_lagna_house")
|
|
for row in skeleton_rows:
|
|
add(_jd(row["start"]), row["start"], "narayana", f"{str(row.get('level')).lower()}_change",
|
|
[sign_house(row.get("sign"))], sign=row.get("sign"), lord=row.get("lord"), level=row.get("level"),
|
|
profile=CONFIG["narayana_skeleton_profile"], basis="natal_lagna_house")
|
|
transits = package.get("slow_transits") or {}
|
|
for event in transits.get("events") or []:
|
|
body = event.get("body")
|
|
if event.get("kind") == "ingress":
|
|
add(_jd(event["instant"]), event["instant"], SYSTEM_OF_BODY[body], "ingress",
|
|
[event.get("to_house_from_lagna")], body=body, to_sign=event.get("to_sign"),
|
|
from_sign=event.get("from_sign"), house_from_moon=event.get("to_house_from_moon"), basis="lagna_house")
|
|
elif event.get("kind") == "station":
|
|
add(_jd(event["instant"]), event["instant"], SYSTEM_OF_BODY[body], "station",
|
|
[sign_house(event.get("sign"))], body=body, motion=event.get("motion"), sign=event.get("sign"),
|
|
basis="lagna_house")
|
|
for contact in transits.get("natal_point_contacts") or []:
|
|
point = contact.get("natal_point")
|
|
house = 1 if point == "Lagna" else houses_of.get(point)
|
|
for crossing in contact.get("crossings") or []:
|
|
add(_jd(crossing["instant"]), crossing["instant"], SYSTEM_OF_BODY[contact["body"]], "natal_contact",
|
|
[house], body=contact["body"], natal_point=point, retrograde=crossing.get("retrograde"),
|
|
repeated=contact.get("repeated"), basis="lagna_house")
|
|
events.sort(key=lambda row: row["jd"])
|
|
for index, event in enumerate(events):
|
|
event["id"] = f"tp-{index + 1:02d}"
|
|
|
|
phases = _phases(package, charts, vim_rows, skeleton_rows, rath_rows, events, functional, natal_houses,
|
|
sign_house, chart_at, range_lo, range_hi, obs_jd, lookback_jd)
|
|
windows = _windows(events, phases, range_lo, range_hi, obs_jd)
|
|
year_overview = [_year_overview(chart) for chart in charts]
|
|
domains = _domain_cards(natal, charts, phases, windows, events)
|
|
conflicts = _conflicts(package, phases, natal)
|
|
return ap.block(
|
|
"computed", "annual_report_facts.build_cards",
|
|
config={key: list(value) if isinstance(value, tuple) else value for key, value in CONFIG.items()},
|
|
house_domains={key: list(value) for key, value in HOUSE_DOMAINS.items()},
|
|
phases=phases, turning_points=[{k: v for k, v in e.items() if k != "jd"} for e in events],
|
|
windows=windows, year_overview=year_overview, domains=domains, conflicts=conflicts,
|
|
)
|
|
|
|
|
|
def _phases(package, charts, vim_rows, skeleton_rows, rath_rows, events, functional, natal_houses, sign_house,
|
|
chart_at, range_lo, range_hi, obs_jd, lookback_jd) -> list[dict]:
|
|
cuts = {range_lo, range_hi}
|
|
for chart in charts:
|
|
cuts.add(_jd(chart["validity"]["start"]))
|
|
for row in vim_rows + skeleton_rows:
|
|
if row.get("level") in ("AD", "PD", "MD"):
|
|
cuts.add(_jd(row["start"]))
|
|
cuts.add(_jd(row["end"]))
|
|
edges = sorted(jd for jd in cuts if jd is not None and range_lo <= jd <= range_hi)
|
|
spans = [[lo, hi] for lo, hi in zip(edges, edges[1:], strict=False) if hi - lo > 1e-6]
|
|
minimum = CONFIG["phase_min_days"]
|
|
merged: list[list[float]] = []
|
|
for span in spans:
|
|
if merged and span[1] - span[0] < minimum or merged and merged[-1][1] - merged[-1][0] < minimum:
|
|
merged[-1][1] = span[1]
|
|
else:
|
|
merged.append(list(span))
|
|
zone = package["report_range"]["start"]
|
|
phases = []
|
|
vim_by_level = {level: [row for row in vim_rows if row.get("level") == level] for level in ("MD", "AD", "PD")}
|
|
nar_by_level = {level: [row for row in skeleton_rows if row.get("level") == level] for level in ("MD", "AD")}
|
|
rath_ad = [row for row in rath_rows if row.get("level") == "AD"]
|
|
for index, (lo, hi) in enumerate(merged, start=1):
|
|
mid = (lo + hi) / 2
|
|
md, ad, pd = (_active(vim_by_level[level], mid) for level in ("MD", "AD", "PD"))
|
|
nmd, nad = (_active(nar_by_level[level], mid) for level in ("MD", "AD"))
|
|
rad = _active(rath_ad, mid)
|
|
chart = chart_at(mid)
|
|
lean_lords = {
|
|
"vimshottari_ad": (ad or {}).get("lord"),
|
|
"vimshottari_pd": (pd or {}).get("lord"),
|
|
"narayana_ad": (nad or {}).get("lord"),
|
|
}
|
|
supports, pressures, seen = [], [], set()
|
|
for key in CONFIG["lean_lords"]:
|
|
planet = lean_lords.get(key)
|
|
if not planet or planet in seen:
|
|
continue
|
|
seen.add(planet)
|
|
role = _role(functional, planet)
|
|
item = {"source": key, "planet": planet, "natal_houses": natal_houses(planet)}
|
|
if role == "support":
|
|
supports.append(item)
|
|
elif role == "pressure":
|
|
pressures.append(item)
|
|
label = ("mixed" if supports and pressures else "smoother" if supports else "tighter" if pressures else "steady")
|
|
annual = []
|
|
if chart:
|
|
blocks = chart["blocks"]
|
|
for row in _overlapping(_rows(blocks["mudda_dasha"], "periods"), lo, hi):
|
|
annual.append({"system": "mudda", "lord": row.get("lord"),
|
|
"annual_house": _annual_house(blocks["chart"], row.get("lord"))})
|
|
for row in _overlapping(_rows(blocks["patyayini"], "periods"), lo, hi):
|
|
annual.append({"system": "patyayini", "lord": row.get("lord"),
|
|
"annual_house": _annual_house(blocks["chart"], row.get("lord"))})
|
|
start_inst = ap.instant(ap._utc_from_jd(lo), _zone_from(zone))
|
|
end_inst = ap.instant(ap._utc_from_jd(hi), _zone_from(zone))
|
|
position = ("earlier" if hi <= lookback_jd else "lookback" if hi <= obs_jd
|
|
else "current" if lo <= obs_jd < hi else "ahead")
|
|
phases.append({
|
|
"id": f"ph-{index:02d}", "start": start_inst, "end": end_inst,
|
|
"duration_days": round(hi - lo, 2), "position": position,
|
|
"year_chart": chart["target_year"] if chart else None,
|
|
"vimshottari": {"md": (md or {}).get("lord"), "ad": (ad or {}).get("lord"), "pd": (pd or {}).get("lord")},
|
|
"narayana": {"profile": CONFIG["narayana_skeleton_profile"],
|
|
"md": {"sign": (nmd or {}).get("sign"), "lord": (nmd or {}).get("lord")},
|
|
"ad": {"sign": (nad or {}).get("sign"), "lord": (nad or {}).get("lord"),
|
|
"house": sign_house((nad or {}).get("sign"))}},
|
|
"narayana_reference": {"profile": "rath", "counts_as_support": CONFIG["narayana_rath_counts_as_support"],
|
|
"ad": {"sign": (rad or {}).get("sign"), "lord": (rad or {}).get("lord")}},
|
|
"lean": {"label": label, "rule": CONFIG["lean_rule"], "supports": supports, "pressures": pressures},
|
|
"annual_unfolding": annual,
|
|
"turning_point_ids": [event["id"] for event in events if lo <= event["jd"] < hi],
|
|
})
|
|
return phases
|
|
|
|
|
|
def _zone_from(boundary: dict):
|
|
return ap._zone(boundary.get("timezone_id"), _offset_hours(boundary.get("utc_offset")))
|
|
|
|
|
|
def _offset_hours(text: Any) -> float:
|
|
value = str(text or "+00:00")
|
|
sign = -1 if value.startswith("-") else 1
|
|
hours, minutes = value.lstrip("+-").split(":")[:2]
|
|
return sign * (int(hours) + int(minutes) / 60)
|
|
|
|
|
|
def _windows(events: list[dict], phases: list[dict], range_lo: float, range_hi: float, obs_jd: float) -> list[dict]:
|
|
clusters: list[list[dict]] = []
|
|
for event in events:
|
|
if event["system"] == "tithi_pravesha" and not CONFIG["tithi_pravesha_in_windows"]:
|
|
continue
|
|
if clusters:
|
|
current = clusters[-1]
|
|
if (event["jd"] - current[-1]["jd"] <= CONFIG["window_cluster_days"]
|
|
and event["jd"] - current[0]["jd"] <= CONFIG["window_max_span_days"]):
|
|
current.append(event)
|
|
continue
|
|
clusters.append([event])
|
|
scored = []
|
|
for cluster in clusters:
|
|
systems = sorted({event["system"] for event in cluster})
|
|
if len(systems) < CONFIG["window_min_independent_systems"]:
|
|
continue
|
|
scored.append((len(systems), len(cluster), cluster, systems))
|
|
scored.sort(key=lambda row: (-row[0], -row[1], row[2][0]["jd"]))
|
|
limit = min(CONFIG["window_target"], CONFIG["window_hard_cap"])
|
|
picked = sorted(scored[:limit], key=lambda row: row[2][0]["jd"])
|
|
windows = []
|
|
widen = CONFIG["unconfirmed_birth_widen_days"]
|
|
for index, (count, _size, cluster, systems) in enumerate(picked, start=1):
|
|
sensitive = any(event["birth_time_sensitive"] for event in cluster)
|
|
lo, hi = cluster[0]["jd"], cluster[-1]["jd"]
|
|
zone = _zone_from(cluster[0]["instant"])
|
|
if sensitive:
|
|
lo, hi = max(lo - widen, range_lo), min(hi + widen, range_hi)
|
|
domain_hits: dict[str, set[str]] = {}
|
|
for event in cluster:
|
|
for domain in event["domains"]:
|
|
domain_hits.setdefault(domain, set()).add(event["system"])
|
|
# Energy (rest and stamina) is the widest bucket; on a tie another domain names the window.
|
|
themes = [domain for domain, _ in sorted(
|
|
domain_hits.items(), key=lambda kv: (-len(kv[1]), kv[0] == "energy", kv[0]))][:2]
|
|
phase = next((p for p in phases if _jd(p["start"]) <= cluster[0]["jd"] < _jd(p["end"])), None)
|
|
limits = []
|
|
if sensitive:
|
|
limits.append({"code": "birth_time_sensitive", "widened_days": widen})
|
|
if phase and phase["lean"]["label"] == "mixed":
|
|
limits.append({"code": "phase_lean_mixed", "phase_id": phase["id"]})
|
|
if "transit_nodes" in systems:
|
|
limits.append({"code": "rahu_ketu_one_axis"})
|
|
windows.append({
|
|
"id": f"w-{index:02d}",
|
|
"status": "candidate",
|
|
"start": ap.instant(ap._utc_from_jd(lo), zone),
|
|
"end": ap.instant(ap._utc_from_jd(hi), zone),
|
|
"independent_systems": systems,
|
|
"independent_count": count,
|
|
"turning_point_ids": [event["id"] for event in cluster],
|
|
"theme_domains": themes,
|
|
"phase_id": phase["id"] if phase else None,
|
|
"phase_lean": phase["lean"]["label"] if phase else None,
|
|
"limits": limits,
|
|
"birth_time_sensitive": sensitive,
|
|
"position": "past" if hi < obs_jd else "current" if lo <= obs_jd else "ahead",
|
|
})
|
|
# The overview's 1-3 things to remember: the strongest windows not yet over.
|
|
ranked = sorted((w for w in windows if w["position"] != "past"),
|
|
key=lambda w: (-w["independent_count"], w["start"]["jd_ut"]))
|
|
for order, window in enumerate(ranked[:3], start=1):
|
|
window["highlight_rank"] = order
|
|
return windows
|
|
|
|
|
|
def _year_overview(chart: dict) -> dict[str, Any]:
|
|
blocks = chart["blocks"]
|
|
chart_block = blocks["chart"]
|
|
year_lord = (blocks["year_lord"].get("data") or {}) if blocks["year_lord"]["status"] == "computed" else {}
|
|
muntha = (blocks["muntha"].get("data") or {}) if blocks["muntha"]["status"] == "computed" else {}
|
|
asc_sign = (chart_block.get("ascendant") or {}).get("sign")
|
|
lord = year_lord.get("year_lord")
|
|
owned = []
|
|
if asc_sign in ap.SIGNS and lord:
|
|
owned = _owned(ap.SIGNS.index(asc_sign), lord)
|
|
strength = blocks["annual_strength"].get("data") or {}
|
|
return {
|
|
"target_year": chart["target_year"],
|
|
"validity": chart["validity"],
|
|
"in_report_range": chart.get("in_report_range"),
|
|
"ascendant": asc_sign,
|
|
"ascendant_lord": chart_block.get("ascendant_lord"),
|
|
"year_lord": {"status": blocks["year_lord"]["status"], "reason": blocks["year_lord"].get("reason"),
|
|
"planet": lord, "annual_house": _annual_house(chart_block, lord), "owned_annual_houses": owned,
|
|
"selection_basis": year_lord.get("selection_basis"),
|
|
"panchavargiya": ((strength.get("panchavargiya_bala") or {}).get(lord) or {}).get("score") if lord else None},
|
|
"muntha": {"status": blocks["muntha"]["status"], "sign": muntha.get("muntha_sign"),
|
|
"annual_house": muntha.get("muntha_house_from_annual_ascendant"),
|
|
"lord": muntha.get("muntha_lord"), "lord_placement": muntha.get("muntha_lord_placement")},
|
|
"mudda_first_lord": blocks["mudda_dasha"].get("starting_lord"),
|
|
}
|
|
|
|
|
|
def _domain_cards(natal: dict, charts: list[dict], phases: list[dict], windows: list[dict], events: list[dict]) -> dict:
|
|
cards = {}
|
|
for domain, houses in HOUSE_DOMAINS.items():
|
|
muntha_years = []
|
|
year_lord_years = []
|
|
for chart in charts:
|
|
overview = _year_overview(chart)
|
|
if overview["muntha"]["annual_house"] in houses:
|
|
muntha_years.append(chart["target_year"])
|
|
if set(overview["year_lord"]["owned_annual_houses"]) & set(houses) or overview["year_lord"]["annual_house"] in houses:
|
|
year_lord_years.append(chart["target_year"])
|
|
ingress_ids = [event["id"] for event in events
|
|
if event["kind"] in ("ingress", "natal_contact", "station") and domain in event["domains"]]
|
|
window_ids = [window["id"] for window in windows if domain in window["theme_domains"]]
|
|
phase_ids = [phase["id"] for phase in phases
|
|
if any(set(item["natal_houses"]) & set(houses) for item in phase["lean"]["supports"] + phase["lean"]["pressures"])]
|
|
signal = "clear" if (window_ids or ingress_ids or muntha_years) else "quiet"
|
|
cards[domain] = {
|
|
"houses": list(houses), "natal": natal["domains"][domain], "signal": signal,
|
|
"muntha_years": muntha_years, "year_lord_years": year_lord_years,
|
|
"transit_turning_point_ids": ingress_ids, "window_ids": window_ids, "phase_ids": phase_ids,
|
|
"health_scope": "rest_and_energy_only" if domain == "energy" else None,
|
|
}
|
|
return cards
|
|
|
|
|
|
def _conflicts(package: dict, phases: list[dict], natal: dict) -> list[dict]:
|
|
out = []
|
|
if not natal["birth_time_confirmed"]:
|
|
out.append({"class": "parameter", "code": "birth_time_not_confirmed",
|
|
"effect": "Lagna houses, D10/D11/D24 and dasha boundaries move with the birth clock; windows widened"})
|
|
for row in package.get("block_index") or []:
|
|
if row["status"] == "blocked":
|
|
out.append({"class": "parameter", "code": "block_blocked", "path": row["path"], "reason": row.get("reason")})
|
|
for phase in phases:
|
|
if phase["lean"]["label"] == "mixed":
|
|
out.append({"class": "opposite_direction", "code": "phase_lean_mixed", "phase_id": phase["id"],
|
|
"supports": [item["planet"] for item in phase["lean"]["supports"]],
|
|
"pressures": [item["planet"] for item in phase["lean"]["pressures"]]})
|
|
ref = phase["narayana_reference"]["ad"]
|
|
main = phase["narayana"]["ad"]
|
|
if ref.get("sign") and main.get("sign") and ref["sign"] != main["sign"]:
|
|
out.append({"class": "parameter", "code": "narayana_profiles_differ", "phase_id": phase["id"],
|
|
"legacy": main["sign"], "rath": ref["sign"], "counted": "legacy only (reference not counted)"})
|
|
return out
|
|
|
|
|
|
def _technique_audit(package: dict, natal: dict) -> list[dict[str, str]]:
|
|
def status_of(block: Any) -> str:
|
|
if not isinstance(block, dict):
|
|
return "blocked"
|
|
return {"computed": "executed"}.get(block.get("status"), block.get("status") or "blocked")
|
|
|
|
charts = package.get("year_charts") or []
|
|
first = charts[-1]["blocks"] if charts and charts[-1].get("blocks") else {}
|
|
functional = natal["functional"]
|
|
rows = [
|
|
{"technique": "Functional Benefic/Malefic", "status": status_of(functional),
|
|
"note": ("功能吉星 " + "、".join(functional.get("benefics", [])) + "；功能凶星 " + "、".join(functional.get("malefics", []))
|
|
+ "；分段的偏顺 / 偏紧只按这一层判断，罗计不单独定")
|
|
if functional.get("status") == "computed" else "未完成，偏顺偏紧不下结论"},
|
|
{"technique": "Vimshottari", "status": status_of(package["natal_periods"].get("vimshottari")),
|
|
"note": "大运 / 小运 / 细运，出生时刻起运"},
|
|
{"technique": "Narayana Dasha", "status": status_of((package["natal_periods"].get("narayana") or {}).get("legacy")),
|
|
"note": "阶段骨架用产品现行算法；Sanjay Rath 版只并列参考，不算第二份支持"},
|
|
{"technique": "Tajika Annual Chart", "status": status_of(first.get("chart")),
|
|
"note": "太阳回归年盘、年主、Muntha，地点为出生地"},
|
|
{"technique": "Mudda Dasha", "status": status_of(first.get("mudda_dasha")), "note": "年度展开，不参与偏顺偏紧投票"},
|
|
{"technique": "Patyayini Dasha", "status": status_of(first.get("patyayini")), "note": "年度展开，固定 365.25 天，只裁不拉伸"},
|
|
{"technique": "Tajika Monthly Charts", "status": status_of(first.get("monthly_charts")), "note": "太阳逐次前进 30°，不进候选窗口"},
|
|
{"technique": "Transits / Sade Sati", "status": status_of(package.get("slow_transits")),
|
|
"note": "木星、土星、罗计换宫、停滞与精确合相；不算容许度相位；Sade Sati 阶段本期未单列"},
|
|
{"technique": "Tithi Pravesha", "status": status_of((package.get("independent_annual_views") or {}).get("tithi_pravesha")),
|
|
"note": "独立年度视角，不并入 Tajika 年，不进候选窗口"},
|
|
{"technique": "Varsha Narayana", "status": "not_applicable", "note": "本仓没有原生实现，未纳入"},
|
|
{"technique": "MEVG / Global Web Evidence", "status": "blocked", "note": "本报告未做外部资料核对，只是计算层归纳"},
|
|
{"technique": "Real Case Calibration", "status": "blocked", "note": "未做独立留出事件验证，不作准确率声明"},
|
|
]
|
|
return rows
|
|
|
|
|
|
def build_annual_report_facts(request: dict[str, Any], *, birth_time_confirmed: bool) -> dict[str, Any]:
|
|
package = ap.build_annual_package(request)
|
|
ctx = ap._Context(request)
|
|
natal = ap._guard("jyotish_engine._compute_chart_from_args", lambda: _natal_base(ctx, birth_time_confirmed))
|
|
if natal.get("status") != "computed":
|
|
cards = ap.block("blocked", "annual_report_facts.build_cards", reason="natal_base_blocked")
|
|
audit: list[dict[str, str]] = []
|
|
else:
|
|
cards = ap._guard("annual_report_facts.build_cards", lambda: build_cards(package, natal))
|
|
audit = _technique_audit(package, natal)
|
|
public_natal = {key: value for key, value in natal.items() if not key.startswith("_")}
|
|
digest = hashlib.sha256(json.dumps(package, sort_keys=True, ensure_ascii=False).encode("utf-8")).hexdigest()
|
|
return ap._plain({
|
|
"schema": SCHEMA,
|
|
"annual_package_version": package["annual_package_version"],
|
|
"package_sha256": digest,
|
|
"parameters": {key: value for key, value in package["parameters"].items() if key != "natal_chart"},
|
|
"report_range": package["report_range"],
|
|
"not_included": package["not_included"],
|
|
"natal_base": public_natal,
|
|
"cards": cards,
|
|
"technique_audit": audit,
|
|
})
|