get_vision_payload

Instructions

Implementation Reference

• src/pythia_oracle_mcp/server.py:1102-1215 (handler)

  The 'get_vision_payload' tool handler function. It fetches the full enriched object for a fired Pythia Vision by ID, including pattern metadata, failure profile, cooldown context, and concurrent fires. Decorated with @mcp.tool() on line 1102.

  @mcp.tool()
  async def get_vision_payload(vision_id: int) -> str:
      """Get the full enriched object for a fired Pythia Vision by id.
  
      Returns the rich AI-facing companion to the on-chain VisionFired event:
      pattern metadata with numeric ranges, failure profile (avg return when
      correct, avg drawdown when wrong, worst drawdown), cooldown context
      (hours since last same-pattern fire on this token, confidence delta vs
      last fire), and concurrent fires from other tokens within the last 24h.
  
      Lightweight on-chain consumers can decode the VisionFired payload bytes
      directly. AI agents reasoning about a specific Vision should use this
      tool — it contains the data needed to size positions and compare against
      historical failure modes, which the on-chain event payload does not.
  
      Args:
          vision_id: integer id of the Vision (returned by get_vision_history)
  
      Returns:
          Multi-section text report. If vision_id is not in the recent window
          (last 20 fires per token), returns a helpful pointer to history.
      """
      data = await _fetch_data()
      visions = data.get("visions", {})
      recent = visions.get("recent", [])
      patterns = visions.get("patterns", [])
  
      found = next((v for v in recent if v.get("id") == vision_id), None)
      if not found:
          return (
              f"Vision id={vision_id} not in recent fires (last 20 per token).\n"
              f"Use get_vision_history(token=...) to find available ids per token."
          )
  
      pattern_meta = next(
          (p for p in patterns
           if p.get("name") == found.get("pattern_name")
           and p.get("token") == found.get("token")),
          None,
      )
  
      out = [
          f"Pythia Vision — id={vision_id}",
          f"Schema version: {visions.get('schema_version', 'v1')}",
          "",
          "─── Fire ───────────────────────────────────────",
          f"  Token:         {found.get('token')}",
          f"  Pattern:       {found.get('pattern_name')} ({hex(found.get('pattern_type', 0)) if isinstance(found.get('pattern_type'), int) else found.get('pattern_type')})",
          f"  Direction:     {found.get('direction')}",
          f"  Confidence:    {found.get('confidence')} / 100",
          f"  Price at fire: ${found.get('price_usd', 0):,.2f}",
          f"  Fired at:      {found.get('fired_at')}",
          f"  Chain:         {found.get('chain')}",
          f"  AI narrative:  {'available' if found.get('haiku_available') else 'mechanical only (AI agent unavailable at fire time)'}",
      ]
  
      # Cooldown / context
      out += ["", "─── Cooldown context ──────────────────────────"]
      cd_same = found.get("cooldown_hours_same_pattern")
      cd_tok = found.get("cooldown_hours_token")
      last_conf = found.get("last_same_pattern_confidence")
      delta = found.get("confidence_delta")
      if cd_same is None and cd_tok is None:
          out.append("  No prior fires for this token — first Vision recorded.")
      else:
          if cd_same is not None:
              out.append(f"  Hours since last same-pattern fire ({found.get('pattern_name')}): {cd_same}")
              if last_conf is not None:
                  out.append(f"  Last same-pattern confidence: {last_conf}  →  delta: {delta:+d}")
          else:
              out.append(f"  First time {found.get('pattern_name')} has fired on {found.get('token')}.")
          if cd_tok is not None and cd_tok != cd_same:
              out.append(f"  Hours since last fire on {found.get('token')} (any pattern): {cd_tok}")
  
      # Concurrent
      concurrent = found.get("concurrent_patterns_24h", []) or []
      out += ["", "─── Concurrent fires (last 24h, other contexts) ──"]
      if not concurrent:
          out.append("  No other fires in the prior 24h — this Vision is idiosyncratic, not market-wide.")
      else:
          for c in concurrent:
              out.append(f"  {c.get('fired_at')}  {c.get('token')} {c.get('pattern_name')}  conf={c.get('confidence')}")
          same_token_count = sum(1 for c in concurrent if c.get("token") == found.get("token"))
          cross_token_count = len(concurrent) - same_token_count
          if cross_token_count >= 2:
              out.append(f"  → {cross_token_count} cross-token fires in last 24h suggests market-wide context.")
  
      # Failure profile
      out += ["", "─── Failure profile (9-year backtest) ─────────"]
      if pattern_meta:
          fp = pattern_meta.get("failure_profile", {})
          out.append(f"  Pattern:                {pattern_meta.get('name')} ({pattern_meta.get('code')})")
          out.append(f"  Data span:              {pattern_meta.get('data_span_years')}")
          out.append(f"  Total fires (backtest): {fp.get('total_fires_backtest', '?')}")
          out.append(f"  Up-rate (24h):          {fp.get('up_rate_pct', '?')}%  (vs {fp.get('base_up_rate_pct', '?')}% baseline → +{fp.get('edge_vs_baseline_pp', '?')}pp edge)")
          out.append(f"  Avg return when correct: {fp.get('avg_return_when_correct_pct', '?'):+}%")
          out.append(f"  Avg return when wrong:   {fp.get('avg_return_when_wrong_pct', '?'):+}%  ← drawdown profile")
          out.append(f"  Worst single drawdown:   {fp.get('worst_drawdown_pct', '?'):+}%")
          out.append(f"  Best single return:      {fp.get('best_return_pct', '?'):+}%")
          out.append(f"  Median 24h return:       {fp.get('median_return_pct', '?'):+}%")
          out.append("")
          ar = pattern_meta.get("accuracy_range_pct", {})
          fv = pattern_meta.get("fold_validation_ratio", {})
          out.append(f"  Accuracy range:         {ar.get('min')}–{ar.get('max')}% (point: {ar.get('point')}%)")
          out.append(f"  Fold validation:        {fv.get('passed')}/{fv.get('total')} ({fv.get('ratio', 0)*100:.0f}%)")
          out.append(f"  Fires/year (estimate):  {pattern_meta.get('fires_per_year_estimate')}")
      else:
          out.append("  Pattern metadata not found in catalog. Use get_visions_info() for catalog.")
  
      out.append("")
      out.append("Note: failure profile is from backtest. Live calibration arrives once vision_outcomes")
      out.append("populates (outcome-tracking cron).")
  
      return "\n".join(out)

• src/pythia_oracle_mcp/server.py:1102-1103 (registration)

  The tool is registered via the @mcp.tool() decorator on the async function get_vision_payload(vision_id: int). The FastMCP instance 'mcp' is created on line 17.

  @mcp.tool()
  async def get_vision_payload(vision_id: int) -> str:

• src/pythia_oracle_mcp/server.py:1102-1123 (schema)

  The input schema is defined by the function signature: vision_id (int). The docstring (lines 1103-1123) describes the return type (multi-section text report) and behavior when vision_id is not found.

  @mcp.tool()
  async def get_vision_payload(vision_id: int) -> str:
      """Get the full enriched object for a fired Pythia Vision by id.
  
      Returns the rich AI-facing companion to the on-chain VisionFired event:
      pattern metadata with numeric ranges, failure profile (avg return when
      correct, avg drawdown when wrong, worst drawdown), cooldown context
      (hours since last same-pattern fire on this token, confidence delta vs
      last fire), and concurrent fires from other tokens within the last 24h.
  
      Lightweight on-chain consumers can decode the VisionFired payload bytes
      directly. AI agents reasoning about a specific Vision should use this
      tool — it contains the data needed to size positions and compare against
      historical failure modes, which the on-chain event payload does not.
  
      Args:
          vision_id: integer id of the Vision (returned by get_vision_history)
  
      Returns:
          Multi-section text report. If vision_id is not in the recent window
          (last 20 fires per token), returns a helpful pointer to history.
      """

• src/pythia_oracle_mcp/server.py:1124-1215 (helper)

  Helper logic: calls _fetch_data() to get live feed-status.json, parses visions.recent and visions.patterns, then assembles a detailed text report with fire details, cooldown context, concurrent fires, and failure profile from backtest data.

  data = await _fetch_data()
  visions = data.get("visions", {})
  recent = visions.get("recent", [])
  patterns = visions.get("patterns", [])
  
  found = next((v for v in recent if v.get("id") == vision_id), None)
  if not found:
      return (
          f"Vision id={vision_id} not in recent fires (last 20 per token).\n"
          f"Use get_vision_history(token=...) to find available ids per token."
      )
  
  pattern_meta = next(
      (p for p in patterns
       if p.get("name") == found.get("pattern_name")
       and p.get("token") == found.get("token")),
      None,
  )
  
  out = [
      f"Pythia Vision — id={vision_id}",
      f"Schema version: {visions.get('schema_version', 'v1')}",
      "",
      "─── Fire ───────────────────────────────────────",
      f"  Token:         {found.get('token')}",
      f"  Pattern:       {found.get('pattern_name')} ({hex(found.get('pattern_type', 0)) if isinstance(found.get('pattern_type'), int) else found.get('pattern_type')})",
      f"  Direction:     {found.get('direction')}",
      f"  Confidence:    {found.get('confidence')} / 100",
      f"  Price at fire: ${found.get('price_usd', 0):,.2f}",
      f"  Fired at:      {found.get('fired_at')}",
      f"  Chain:         {found.get('chain')}",
      f"  AI narrative:  {'available' if found.get('haiku_available') else 'mechanical only (AI agent unavailable at fire time)'}",
  ]
  
  # Cooldown / context
  out += ["", "─── Cooldown context ──────────────────────────"]
  cd_same = found.get("cooldown_hours_same_pattern")
  cd_tok = found.get("cooldown_hours_token")
  last_conf = found.get("last_same_pattern_confidence")
  delta = found.get("confidence_delta")
  if cd_same is None and cd_tok is None:
      out.append("  No prior fires for this token — first Vision recorded.")
  else:
      if cd_same is not None:
          out.append(f"  Hours since last same-pattern fire ({found.get('pattern_name')}): {cd_same}")
          if last_conf is not None:
              out.append(f"  Last same-pattern confidence: {last_conf}  →  delta: {delta:+d}")
      else:
          out.append(f"  First time {found.get('pattern_name')} has fired on {found.get('token')}.")
      if cd_tok is not None and cd_tok != cd_same:
          out.append(f"  Hours since last fire on {found.get('token')} (any pattern): {cd_tok}")
  
  # Concurrent
  concurrent = found.get("concurrent_patterns_24h", []) or []
  out += ["", "─── Concurrent fires (last 24h, other contexts) ──"]
  if not concurrent:
      out.append("  No other fires in the prior 24h — this Vision is idiosyncratic, not market-wide.")
  else:
      for c in concurrent:
          out.append(f"  {c.get('fired_at')}  {c.get('token')} {c.get('pattern_name')}  conf={c.get('confidence')}")
      same_token_count = sum(1 for c in concurrent if c.get("token") == found.get("token"))
      cross_token_count = len(concurrent) - same_token_count
      if cross_token_count >= 2:
          out.append(f"  → {cross_token_count} cross-token fires in last 24h suggests market-wide context.")
  
  # Failure profile
  out += ["", "─── Failure profile (9-year backtest) ─────────"]
  if pattern_meta:
      fp = pattern_meta.get("failure_profile", {})
      out.append(f"  Pattern:                {pattern_meta.get('name')} ({pattern_meta.get('code')})")
      out.append(f"  Data span:              {pattern_meta.get('data_span_years')}")
      out.append(f"  Total fires (backtest): {fp.get('total_fires_backtest', '?')}")
      out.append(f"  Up-rate (24h):          {fp.get('up_rate_pct', '?')}%  (vs {fp.get('base_up_rate_pct', '?')}% baseline → +{fp.get('edge_vs_baseline_pp', '?')}pp edge)")
      out.append(f"  Avg return when correct: {fp.get('avg_return_when_correct_pct', '?'):+}%")
      out.append(f"  Avg return when wrong:   {fp.get('avg_return_when_wrong_pct', '?'):+}%  ← drawdown profile")
      out.append(f"  Worst single drawdown:   {fp.get('worst_drawdown_pct', '?'):+}%")
      out.append(f"  Best single return:      {fp.get('best_return_pct', '?'):+}%")
      out.append(f"  Median 24h return:       {fp.get('median_return_pct', '?'):+}%")
      out.append("")
      ar = pattern_meta.get("accuracy_range_pct", {})
      fv = pattern_meta.get("fold_validation_ratio", {})
      out.append(f"  Accuracy range:         {ar.get('min')}–{ar.get('max')}% (point: {ar.get('point')}%)")
      out.append(f"  Fold validation:        {fv.get('passed')}/{fv.get('total')} ({fv.get('ratio', 0)*100:.0f}%)")
      out.append(f"  Fires/year (estimate):  {pattern_meta.get('fires_per_year_estimate')}")
  else:
      out.append("  Pattern metadata not found in catalog. Use get_visions_info() for catalog.")
  
  out.append("")
  out.append("Note: failure profile is from backtest. Live calibration arrives once vision_outcomes")
  out.append("populates (outcome-tracking cron).")
  
  return "\n".join(out)