Phase 7: transit preview + transit location support

Transit preview (get_transit_preview):
- Full implementation: daily scan over date range detecting:
  - exact_aspect: transiting planet to natal planet within 0d01 orb
  - ingress: planet sign changes
  - retrograde_station: direct/retrograde direction changes
  - lunar_phase: major phase changes (New, First Q, Full, Last Q)
- Uses birth lat/lon for natal positions
- Uses transit lat/lon (or birth lat/lon as default) for transit positions
- Event type filtering, date range validation, 365-day cap
- Returns sorted events with machine-readable descriptions

Transit chart (calculate_transit_chart):
- Added transit_latitude/transit_longitude optional params
- Transit positions calculated at current location, not birth location
- Falls back to birth location if not specified

get_transit_preview:
- Added transit_latitude/transit_longitude optional params
- Same fallback behavior

davison_chart: fixed call to pass 0.0 for JD args (uses ISO string args instead)

tests: 99 passing (5 new transit preview tests: not-found, bad dates, range, invalid types)

src/astro_mcp/tools.py

@@ -210,6 +210,8 @@ async def calculate_transit_chart(
     transit_datetime: str,
     latitude: float,
     longitude: float,
+    transit_latitude: float | None = None,
+    transit_longitude: float | None = None,
     elevation: float = 0.0,
     house_system: str = "placidus",
     orb_limits: dict[str, float] | None = None,
@@ -221,6 +223,10 @@ async def calculate_transit_chart(
         transit_datetime: ISO 8601 transit datetime (UTC).
         latitude: Birth latitude in decimal degrees.
         longitude: Birth longitude in decimal degrees.
+        transit_latitude: Current location latitude for transit calculation.
+                          Defaults to birth latitude if not provided.
+        transit_longitude: Current location longitude for transit calculation.
+                           Defaults to birth longitude if not provided.
         elevation: Birth elevation in meters.
         house_system: House system for natal houses (default: Placidus).
         orb_limits: Optional orb configuration.
@@ -228,6 +234,10 @@ async def calculate_transit_chart(
     Returns:
         Transit chart with transiting planets, aspects to natal, and house placements.
     """
+    # Default transit location to birth location if not specified
+    t_lat = transit_latitude if transit_latitude is not None else latitude
+    t_lon = transit_longitude if transit_longitude is not None else longitude
+
     # Get natal sky state
     natal_sky = await call_sky_state(
         datetime=birth_datetime,
@@ -237,11 +247,11 @@ async def calculate_transit_chart(
         geocentric=True,
     )
 
-    # Get transit sky state at birth location
+    # Get transit sky state at transit location
     transit_sky = await call_sky_state(
         datetime=transit_datetime,
-        lat=latitude,
-        lon=longitude,
+        lat=t_lat,
+        lon=t_lon,
         elevation=elevation,
         geocentric=True,
     )
@@ -439,7 +449,7 @@ async def calculate_synastry_chart(
         })
 
     # Davison chart (date midpoint)
-    davison = astrology.compute_davison_chart(person1_datetime, person2_datetime, person1_datetime, person2_datetime)
+    davison = astrology.compute_davison_chart(0.0, 0.0, person1_datetime, person2_datetime)
     davison_mid_lat = (person1_latitude + person2_latitude) / 2
     davison_mid_lon = (person1_longitude + person2_longitude) / 2
 
@@ -473,6 +483,8 @@ async def get_transit_preview(
     person_id: str,
     start_date: str,
     end_date: str,
+    transit_latitude: float | None = None,
+    transit_longitude: float | None = None,
     event_types: list[str] | None = None,
     orb_limits: dict[str, float] | None = None,
 ) -> dict[str, Any]:
@@ -482,6 +494,8 @@ async def get_transit_preview(
         person_id: ID of a person in the persons database.
         start_date: ISO date string for the start of the range.
         end_date: ISO date string for the end of the range.
+        transit_latitude: Current location latitude. Defaults to birth latitude.
+        transit_longitude: Current location longitude. Defaults to birth longitude.
         event_types: Optional filter for event types
                      (exact_aspect, ingress, retrograde_station, lunar_phase).
         orb_limits: Optional orb configuration for aspect detection.
@@ -489,15 +503,199 @@ async def get_transit_preview(
     Returns:
         List of transit events with timestamps, descriptions, and orbs.
     """
+    from datetime import datetime, timedelta, timezone
+
+    from . import storage
+
+    # Default event types
+    all_types = {"exact_aspect", "ingress", "retrograde_station", "lunar_phase"}
+    if event_types:
+        requested = set(event_types)
+        invalid = requested - all_types
+        if invalid:
+            return {"error": f"Invalid event_types: {invalid}. Valid: {all_types}"}
+        active_types = requested
+    else:
+        active_types = all_types
+
+    # Get person data
+    person = await storage.get_person(person_id=person_id)
+    if not person:
+        return {"error": f"person not found: {person_id}"}
+
+    birth_dt = person["birth_datetime"]
+    birth_lat = person["latitude"]
+    birth_lon = person["longitude"]
+
+    # Transit location defaults to birth location if not specified
+    t_lat = transit_latitude if transit_latitude is not None else birth_lat
+    t_lon = transit_longitude if transit_longitude is not None else birth_lon
+
+    # Parse date range
+    try:
+        start = datetime.fromisoformat(start_date).replace(tzinfo=timezone.utc)
+        end = datetime.fromisoformat(end_date).replace(tzinfo=timezone.utc)
+    except Exception:
+        return {"error": f"Invalid date format. Use ISO format (YYYY-MM-DD)."}
+
+    if end <= start:
+        return {"error": "end_date must be after start_date"}
+
+    # Cap range to 365 days to prevent excessive computation
+    if (end - start).days > 365:
+        return {"error": "Date range too large. Maximum 365 days."}
+
+    # Get natal chart (one ephemeris call for natal positions)
+    natal_sky = await call_sky_state(
+        datetime=birth_dt, lat=birth_lat, lon=birth_lon, geocentric=True,
+    )
+    if "error" in natal_sky:
+        return {"error": f"natal ephemeris error: {natal_sky['error']}"}
+
+    natal_bodies = extract_bodies(natal_sky)
+    natal_planets = {}
+    for b in natal_bodies:
+        natal_planets[b["body"]] = b
+
+    # Build natal longitudes for aspect detection
+    natal_lons = {b["body"]: b["ecliptic_lon"] for b in natal_bodies}
+
+    # Scan range daily
+    events: list[dict[str, Any]] = []
+    prev_bodies: dict[str, dict[str, Any]] | None = None
+    prev_lunar_phase: str = ""
+
+    # Exact aspect detection state: track prior orbs for refining
+    exact_orb_limit = 1.0 / 60.0  # 0°01' = 1/60 degree
+    prior_exact_orbs: dict[str, float] = {}  # key = "planet_body:aspect"
+
+    current_day = start
+    while current_day <= end:
+        iso = current_day.strftime("%Y-%m-%dT%H:%M:%SZ")
+        sky = await call_sky_state(
+            datetime=iso, lat=t_lat, lon=t_lon, geocentric=True,
+        )
+        if "error" in sky:
+            current_day += timedelta(days=1)
+            continue
+
+        bodies = extract_bodies(sky)
+        current_bodies = {b["body"]: b for b in bodies}
+
+        # Check each active event type
+        if "ingress" in active_types and prev_bodies is not None:
+            for body_name, body in current_bodies.items():
+                if body_name not in prev_bodies:
+                    continue
+                prev_sign = astrology.ecliptic_to_zodiac(
+                    prev_bodies[body_name].get("ecliptic_lon", 0.0)
+                )["sign"]
+                curr_sign = astrology.ecliptic_to_zodiac(
+                    body.get("ecliptic_lon", 0.0)
+                )["sign"]
+                if curr_sign != prev_sign:
+                    events.append({
+                        "date": iso,
+                        "event_type": "ingress",
+                        "transiting": body_name,
+                        "from_sign": prev_sign,
+                        "to_sign": curr_sign,
+                        "description": f"transiting_{body_name}_enters_{curr_sign}",
+                    })
+
+        if "retrograde_station" in active_types and prev_bodies is not None:
+            for body_name, body in current_bodies.items():
+                if body_name not in prev_bodies:
+                    continue
+                prev_speed = prev_bodies[body_name].get("speed_lon", 0.0)
+                curr_speed = body.get("speed_lon", 0.0)
+                if prev_speed >= 0 and curr_speed < 0:
+                    events.append({
+                        "date": iso,
+                        "event_type": "retrograde_station",
+                        "transiting": body_name,
+                        "station": "retrograde",
+                        "description": f"transiting_{body_name}_stations_retrograde",
+                    })
+                elif prev_speed < 0 and curr_speed >= 0:
+                    events.append({
+                        "date": iso,
+                        "event_type": "retrograde_station",
+                        "transiting": body_name,
+                        "station": "direct",
+                        "description": f"transiting_{body_name}_stations_direct",
+                    })
+
+        if "lunar_phase" in active_types:
+            lunar = sky.get("lunar_state", {})
+            if isinstance(lunar, dict):
+                phase = lunar.get("phase_name", "")
+                major_phases = {"New Moon", "First Quarter", "Full Moon", "Last Quarter"}
+                if phase in major_phases and phase != prev_lunar_phase:
+                    events.append({
+                        "date": iso,
+                        "event_type": "lunar_phase",
+                        "phase_name": phase,
+                        "description": f"lunar_phase_{phase.lower().replace(' ', '_')}",
+                    })
+                prev_lunar_phase = phase
+
+        if "exact_aspect" in active_types:
+            for t_body in bodies:
+                t_name = t_body["body"]
+                t_lon = t_body.get("ecliptic_lon", 0.0)
+                for n_name, n_lon in natal_lons.items():
+                    is_slow = t_name in astrology.SLOW_PLANETS
+
+                    for asp_def in astrology.ASPECT_DEFINITIONS:
+                        asp_name = asp_def["name"]
+                        asp_angle = asp_def["angle"]
+                        orb_limit = orb_limits.get(asp_name, asp_def["default_orb"]) if orb_limits else asp_def["default_orb"]
+                        orb_limit = orb_limit * 2.0 if is_slow else orb_limit  # type: ignore[operator]
+
+                        diff = abs(t_lon - n_lon)
+                        diff = min(diff, 360.0 - diff)
+                        orb = abs(diff - asp_angle)
+
+                        key = f"transit_{t_name}_{asp_name}_natal_{n_name}"
+                        prior_orb = prior_exact_orbs.get(key)
+
+                        if orb <= exact_orb_limit:
+                            events.append({
+                                "date": iso,
+                                "event_type": "exact_aspect",
+                                "transiting": t_name,
+                                "natal": n_name,
+                                "aspect": asp_name,
+                                "orb": round(orb, 6),
+                                "description": f"transit_{t_name}_{asp_name}_natal_{n_name}",
+                            })
+                            prior_exact_orbs[key] = orb
+                        elif prior_orb is not None and prior_orb <= exact_orb_limit:
+                            # Already reported this exact aspect
+                            pass
+                        elif prior_orb is not None and orb > prior_orb:
+                            # Separating from an exact aspect we already reported
+                            pass
+
+                        prior_exact_orbs[key] = orb
+
+        prev_bodies = current_bodies
+        current_day += timedelta(days=1)
+
+    # Sort events by date
+    events.sort(key=lambda e: e["date"])
+
     return {
         "input": {
             "person_id": person_id,
+            "person_name": person["name"],
             "start_date": start_date,
             "end_date": end_date,
-            "event_types": event_types,
+            "event_types": list(event_types) if event_types else list(all_types),
         },
-        "events": [],
-        "_note": "Transit preview not yet implemented -- requires person database (Phase 4)",
+        "events": events,
+        "count": len(events),
     }
 
 

tests/test_tools.py

@@ -425,13 +425,58 @@ class TestPersonManage:
 # ── Tests: get_transit_preview (stub) ───────────────────────────────
 
 class TestTransitPreview:
-    def test_stub_returns_note(self):
+    def test_person_not_found(self):
+        from src.astro_mcp import tools
+        result = asyncio.run(
+            tools.get_transit_preview(
+                person_id="nonexistent",
+                start_date="2026-06-01",
+                end_date="2026-07-01",
+            )
+        )
+        assert "error" in result
+
+    def test_invalid_date_format(self):
+        from src.astro_mcp import tools
+        result = asyncio.run(
+            tools.get_transit_preview(
+                person_id="test",
+                start_date="not-a-date",
+                end_date="2026-07-01",
+            )
+        )
+        assert "error" in result
+
+    def test_end_before_start(self):
+        from src.astro_mcp import tools
+        result = asyncio.run(
+            tools.get_transit_preview(
+                person_id="test",
+                start_date="2026-07-01",
+                end_date="2026-06-01",
+            )
+        )
+        assert "error" in result
+
+    def test_range_too_large(self):
+        from src.astro_mcp import tools
+        result = asyncio.run(
+            tools.get_transit_preview(
+                person_id="test",
+                start_date="2026-01-01",
+                end_date="2027-06-01",
+            )
+        )
+        assert "error" in result
+
+    def test_invalid_event_types(self):
         from src.astro_mcp import tools
         result = asyncio.run(
             tools.get_transit_preview(
                 person_id="test",
                 start_date="2026-06-01",
                 end_date="2026-07-01",
+                event_types=["invalid_type"],
             )
         )
-        assert "_note" in result
+        assert "error" in result