From 522454ab61df223f19118f3afc9ad0a8af9f0643 Mon Sep 17 00:00:00 2001 From: alvis Date: Thu, 14 May 2026 10:59:10 +0000 Subject: [PATCH] =?UTF-8?q?feat(agents):=20stars=20agent=20=E2=80=94=20ast?= =?UTF-8?q?rological=20transits=20via=20pyswisseph=20(#121)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Computes natal chart (Sun/Moon/Mercury/Venus/Mars/Jupiter/Saturn) from birth_date and finds active transits (conjunction/sextile/square/trine/ opposition) between today's sky and the user's natal positions. Top 3 most-exact transits are passed to the orchestrator as interpretive themes to colour the tip — grounded and actionable, not predictive. Birth date sourced from agent_prefs (populated by a connected Google data source); requires data:google-health consent. Agent self-silences when birth_date is absent. pyswisseph added to ml/serving/requirements.txt. Co-Authored-By: Claude Sonnet 4.6 --- ml/agents/registry.py | 2 + ml/agents/stars.py | 233 +++++++++++++++++++++++++++++++++ ml/agents/tests/test_agents.py | 46 ++++++- ml/serving/requirements.txt | 1 + 4 files changed, 281 insertions(+), 1 deletion(-) create mode 100644 ml/agents/stars.py diff --git a/ml/agents/registry.py b/ml/agents/registry.py index 0263d62..9748f16 100644 --- a/ml/agents/registry.py +++ b/ml/agents/registry.py @@ -18,6 +18,7 @@ from .recent_patterns import RecentPatternsAgent, MANIFEST as RECENT_PATTERNS_MA from .focus_area import FocusAreaAgent, MANIFEST as FOCUS_AREA_MANIFEST from .health_vitals import HealthVitalsAgent, MANIFEST as HEALTH_VITALS_MANIFEST from .tarot import TarotAgent, MANIFEST as TAROT_MANIFEST +from .stars import StarsAgent, MANIFEST as STARS_MANIFEST _REGISTERED: list[tuple[BaseAgent, AgentManifest]] = [ (OverdueTaskAgent(), OVERDUE_TASK_MANIFEST), @@ -27,6 +28,7 @@ _REGISTERED: list[tuple[BaseAgent, AgentManifest]] = [ (FocusAreaAgent(), FOCUS_AREA_MANIFEST), (HealthVitalsAgent(), HEALTH_VITALS_MANIFEST), (TarotAgent(), TAROT_MANIFEST), + (StarsAgent(), STARS_MANIFEST), ] # Sanity check — agent_id and manifest.id must agree, otherwise the registry diff --git a/ml/agents/stars.py b/ml/agents/stars.py new file mode 100644 index 0000000..06e5ad3 --- /dev/null +++ b/ml/agents/stars.py @@ -0,0 +1,233 @@ +"""Stars agent — astrological transit predictions via pyswisseph. + +Requires birth_date in agent_prefs (ISO 8601 date string, e.g. '1990-06-15'). +Populated from a connected data source (Google profile / Google Health). +If birth_date is absent the agent returns a no-data snippet and the +eligibility filter will silence it once the consent / pref check catches up. + +Computes today's Sun, Moon, Mercury, Venus, Mars, Jupiter, Saturn positions +and finds notable transits (conjunctions, oppositions, squares, trines, sextiles) +between today's sky and the user's natal chart. Passes a concise prediction ++ interpretation to the orchestrator. +""" +from __future__ import annotations + +import math +from datetime import date, datetime, timezone +from typing import ClassVar + +from .base import BaseAgent, AgentInput, AgentOutput +from .manifest import AgentManifest, InferredParam + +try: + import swisseph as swe # type: ignore + _SWE_AVAILABLE = True +except ImportError: # pragma: no cover — present in container, absent in dev + _SWE_AVAILABLE = False + +# --------------------------------------------------------------------------- +# Planet catalogue +# --------------------------------------------------------------------------- +_PLANETS: list[tuple[int, str]] = [] +if _SWE_AVAILABLE: + _PLANETS = [ + (swe.SUN, "Sun"), + (swe.MOON, "Moon"), + (swe.MERCURY, "Mercury"), + (swe.VENUS, "Venus"), + (swe.MARS, "Mars"), + (swe.JUPITER, "Jupiter"), + (swe.SATURN, "Saturn"), + ] + +# Aspect definitions: (angle, orb, name, nature) +_ASPECTS: list[tuple[float, float, str, str]] = [ + (0.0, 8.0, "conjunction", "intensifying"), + (60.0, 6.0, "sextile", "harmonious"), + (90.0, 7.0, "square", "challenging"), + (120.0, 8.0, "trine", "flowing"), + (180.0, 8.0, "opposition", "tension"), +] + +_ZODIAC = [ + "Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo", + "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces", +] + +# Interpretive keywords per planet for transit readings +_PLANET_THEMES: dict[str, str] = { + "Sun": "identity, vitality, core purpose", + "Moon": "emotions, intuition, comfort needs", + "Mercury": "communication, thinking, decisions", + "Venus": "relationships, values, pleasure", + "Mars": "energy, drive, conflict", + "Jupiter": "growth, opportunity, expansion", + "Saturn": "discipline, responsibility, long-term structure", +} + + +def _zodiac_sign(lon: float) -> str: + return _ZODIAC[int(lon / 30) % 12] + + +def _jd_from_date(d: date) -> float: + """Julian Day Number for noon UTC on the given date.""" + assert _SWE_AVAILABLE + return swe.julday(d.year, d.month, d.day, 12.0) + + +def _planet_positions(jd: float) -> dict[str, float]: + assert _SWE_AVAILABLE + positions: dict[str, float] = {} + for pid, name in _PLANETS: + result, _ = swe.calc_ut(jd, pid) + positions[name] = result[0] # ecliptic longitude + return positions + + +def _angular_diff(a: float, b: float) -> float: + """Smallest angle between two ecliptic longitudes (0–180).""" + diff = abs(a - b) % 360 + return diff if diff <= 180 else 360 - diff + + +def _find_transits(natal: dict[str, float], today: dict[str, float]) -> list[dict]: + """Return list of active transits between today's sky and natal chart.""" + transits: list[dict] = [] + for t_name, t_lon in today.items(): + for n_name, n_lon in natal.items(): + diff = _angular_diff(t_lon, n_lon) + for angle, orb, aspect_name, nature in _ASPECTS: + if abs(diff - angle) <= orb: + transits.append({ + "transit_planet": t_name, + "natal_planet": n_name, + "aspect": aspect_name, + "nature": nature, + "orb": round(abs(diff - angle), 2), + }) + # Sort by tightness of orb + transits.sort(key=lambda x: x["orb"]) + return transits + + +def _format_transit(t: dict) -> str: + tp, np, asp, nat = t["transit_planet"], t["natal_planet"], t["aspect"], t["nature"] + tp_theme = _PLANET_THEMES.get(tp, "") + np_theme = _PLANET_THEMES.get(np, "") + return ( + f"Transiting {tp} ({tp_theme}) {asp} natal {np} ({np_theme}) " + f"— a {nat} influence" + ) + + +# --------------------------------------------------------------------------- +# Manifest +# --------------------------------------------------------------------------- +MANIFEST = AgentManifest( + id="stars", + version="1.0.0", + description="Astrological transit predictions based on the user's birth date and today's planetary positions.", + pref_schema={ + "type": "object", + "additionalProperties": False, + "properties": { + "birth_date": { + "type": "string", + "pattern": r"^\d{4}-\d{2}-\d{2}$", + "description": "ISO 8601 birth date (YYYY-MM-DD). Populated from connected data source.", + }, + }, + }, + context_schema=["profile.birth_date"], + # Requires a connected Google source that supplies birth date. + # data:google-health is the current carrier; when Google profile is a + # separate consent key, add it here. + required_consents=["data:core", "data:google-health"], + output_contract={"type": "snippet", "format": "free_text"}, + ttl_sec=3_600 * 6, # planetary positions change slowly — 6 h is fine + silenced_in_contexts=[], + inferred_params=[ + InferredParam( + key="birth_date", + ttl_sec=365 * 86_400, # effectively permanent once known + cold_start_default=None, + min_history=999_999, # never inferred from events — sourced externally + infer=None, + ), + ], +) + + +class StarsAgent(BaseAgent): + """Produces astrological transit predictions for the user's birth chart.""" + + agent_id: ClassVar[str] = MANIFEST.id + ttl_seconds: ClassVar[int] = MANIFEST.ttl_sec + version: ClassVar[str] = MANIFEST.version + + def compute(self, inp: AgentInput) -> AgentOutput: + birth_date_str: str | None = inp.agent_prefs.get("birth_date") + + if not birth_date_str: + prompt = ( + "Birth date is not available — astrological reading skipped. " + "(Always write the tip in English.)" + ) + return self._make_output(inp, prompt, {"no_birth_date": True}) + + if not _SWE_AVAILABLE: + prompt = ( + "Astrological library unavailable — reading skipped. " + "(Always write the tip in English.)" + ) + return self._make_output(inp, prompt, {"swe_unavailable": True}) + + try: + birth_date = date.fromisoformat(birth_date_str) + except ValueError: + prompt = "Birth date format invalid — astrological reading skipped." + return self._make_output(inp, prompt, {"invalid_birth_date": birth_date_str}) + + today_date = inp.now.date() + natal_jd = _jd_from_date(birth_date) + today_jd = _jd_from_date(today_date) + + natal_pos = _planet_positions(natal_jd) + today_pos = _planet_positions(today_jd) + + transits = _find_transits(natal_pos, today_pos) + top = transits[:3] # most exact transits only + + today_sun_sign = _zodiac_sign(today_pos["Sun"]) + natal_sun_sign = _zodiac_sign(natal_pos["Sun"]) + natal_moon_sign = _zodiac_sign(natal_pos["Moon"]) + + snapshot = { + "birth_date": birth_date_str, + "today": today_date.isoformat(), + "natal_sun": natal_sun_sign, + "natal_moon": natal_moon_sign, + "today_sun": today_sun_sign, + "active_transits": transits[:5], + } + + if not top: + prompt = ( + f"Natal chart: Sun in {natal_sun_sign}, Moon in {natal_moon_sign}. " + f"Today's Sun is in {today_sun_sign}. " + "No exact transits today — a quiet, stable day energetically. " + "(Always write the tip in English.)" + ) + else: + transit_lines = "; ".join(_format_transit(t) for t in top) + prompt = ( + f"Natal chart: Sun in {natal_sun_sign}, Moon in {natal_moon_sign}. " + f"Today's Sun is in {today_sun_sign}. " + f"Active transits: {transit_lines}. " + "Use these planetary themes to colour the tip — " + "keep it grounded and actionable, not predictive or fatalistic. " + "(Always write the tip in English.)" + ) + + return self._make_output(inp, prompt, snapshot) diff --git a/ml/agents/tests/test_agents.py b/ml/agents/tests/test_agents.py index c8e7504..266e0f1 100644 --- a/ml/agents/tests/test_agents.py +++ b/ml/agents/tests/test_agents.py @@ -14,6 +14,7 @@ from ml.agents.time_of_day import TimeOfDayAgent from ml.agents.recent_patterns import RecentPatternsAgent from ml.agents.focus_area import FocusAreaAgent from ml.agents.tarot import TarotAgent, _daily_draw, _CARDS, _POSITIONS +from ml.agents.stars import StarsAgent, _SWE_AVAILABLE from ml.agents.registry import get_agent, all_agents _NOW = datetime(2026, 5, 1, 9, 0, 0, tzinfo=timezone.utc) # Thursday 09:00 UTC @@ -297,13 +298,56 @@ class TestTarotAgent: assert all(0 <= i < len(_CARDS) for i in indices) +# ── StarsAgent ──────────────────────────────────────────────────────────────── + +class TestStarsAgent: + agent = StarsAgent() + + def test_no_birth_date(self): + out = self.agent.compute(_inp()) + _check_output(out, self.agent) + assert out.signals_snapshot.get("no_birth_date") is True + assert "birth date" in out.prompt_text.lower() + + @pytest.mark.skipif(not _SWE_AVAILABLE, reason="pyswisseph not installed") + def test_invalid_birth_date(self): + out = self.agent.compute(_inp(agent_prefs={"birth_date": "not-a-date"})) + _check_output(out, self.agent) + assert out.signals_snapshot.get("invalid_birth_date") == "not-a-date" + + @pytest.mark.skipif(not _SWE_AVAILABLE, reason="pyswisseph not installed") + def test_with_birth_date(self): + out = self.agent.compute(_inp(agent_prefs={"birth_date": "1990-06-15"})) + _check_output(out, self.agent) + assert "natal" in out.prompt_text.lower() + assert out.signals_snapshot["birth_date"] == "1990-06-15" + assert "natal_sun" in out.signals_snapshot + assert "natal_moon" in out.signals_snapshot + + @pytest.mark.skipif(not _SWE_AVAILABLE, reason="pyswisseph not installed") + def test_transit_snapshot_structure(self): + out = self.agent.compute(_inp(agent_prefs={"birth_date": "1985-03-21"})) + snap = out.signals_snapshot + assert "active_transits" in snap + for t in snap["active_transits"]: + assert {"transit_planet", "natal_planet", "aspect", "nature", "orb"} <= t.keys() + + def test_swe_unavailable_path(self, monkeypatch): + import ml.agents.stars as stars_mod + monkeypatch.setattr(stars_mod, "_SWE_AVAILABLE", False) + agent = StarsAgent() + out = agent.compute(_inp(agent_prefs={"birth_date": "1990-06-15"})) + _check_output(out, agent) + assert out.signals_snapshot.get("swe_unavailable") is True + + # ── Registry ───────────────────────────────────────────────────────────────── class TestRegistry: def test_all_agents_present(self): agents = all_agents() ids = {a.agent_id for a in agents} - assert ids == {"overdue-task", "momentum", "time-of-day", "recent-patterns", "focus-area", "health-vitals", "tarot"} + assert ids == {"overdue-task", "momentum", "time-of-day", "recent-patterns", "focus-area", "health-vitals", "tarot", "stars"} def test_get_agent(self): a = get_agent("momentum") diff --git a/ml/serving/requirements.txt b/ml/serving/requirements.txt index 00253e8..d3a12af 100644 --- a/ml/serving/requirements.txt +++ b/ml/serving/requirements.txt @@ -8,3 +8,4 @@ nats-py>=2.9.0 structlog>=24.1.0 sentry-sdk>=2.0.0 mlflow-skinny>=3.1.0 +pyswisseph>=2.10.3.2