| """The nine public tools. Each returns JSON text. No orders and no state/ writes.""" |
|
|
| from __future__ import annotations |
|
|
| import csv |
| import json |
| import math |
| import os |
| import tempfile |
| from pathlib import Path |
| from typing import Any |
|
|
| from aureon_public import paths |
| from aureon_public.guard_loader import aureon_source |
| from aureon_public.quiet import quiet |
| from aureon_public.text import AURIS_FOUR, HOLD_LINE |
|
|
| _AURIS_FIELDS = ("volatility", "momentum", "volume", "spread") |
|
|
|
|
| def _plain(value: Any) -> Any: |
| if isinstance(value, dict): |
| return {str(key): _plain(item) for key, item in value.items()} |
| if isinstance(value, (list, tuple)): |
| return [_plain(item) for item in value] |
| if isinstance(value, bool) or value is None or isinstance(value, str): |
| return value |
| if isinstance(value, int): |
| return int(value) |
| if isinstance(value, float): |
| return float(value) |
| item = getattr(value, "item", None) |
| if callable(item): |
| try: |
| return _plain(item()) |
| except Exception: |
| pass |
| return str(value) |
|
|
|
|
| _OMITTED_PUBLIC_KEYS = ("consciousness_level", "consciousness_psi") |
|
|
|
|
| def _omit_public_keys(body: dict[str, Any]) -> dict[str, Any]: |
| for key in _OMITTED_PUBLIC_KEYS: |
| body.pop(key, None) |
| return body |
|
|
|
|
| def _strip_consciousness(value: Any) -> Any: |
| if isinstance(value, dict): |
| return { |
| str(key): _strip_consciousness(item) |
| for key, item in value.items() |
| if not str(key).startswith("consciousness_") |
| } |
| if isinstance(value, list): |
| return [_strip_consciousness(item) for item in value] |
| return value |
|
|
|
|
| def _dumps(payload: dict[str, Any]) -> str: |
| body = _strip_consciousness(_plain(payload)) |
| body["aureon_source"] = aureon_source() |
| return json.dumps(body, ensure_ascii=False, allow_nan=False, sort_keys=True) |
|
|
|
|
| def _load(path: Path) -> dict[str, Any]: |
| loaded = json.loads(path.read_text(encoding="utf-8")) |
| if not isinstance(loaded, dict): |
| raise TypeError(path.name) |
| return loaded |
|
|
|
|
| def _finite(value: object) -> float | None: |
| if isinstance(value, bool) or value is None: |
| return None |
| if not isinstance(value, (int, float)): |
| return None |
| number = float(value) |
| if not math.isfinite(number): |
| return None |
| return number |
|
|
|
|
| def _brain() -> dict[str, Any]: |
| return _load(paths.brain_json()) |
|
|
|
|
| def _path_block(brain: dict[str, Any], enable_multiverse: bool) -> dict[str, Any]: |
| hnc = brain.get("hnc") |
| if not isinstance(hnc, dict): |
| raise KeyError("hnc") |
| if enable_multiverse: |
| block = hnc.get("multiverse_extension") |
| else: |
| block = hnc |
| if not isinstance(block, dict): |
| raise KeyError("brain path") |
| return block |
|
|
|
|
| def _cap(enable_multiverse: bool) -> int: |
| steps = _path_block(_brain(), enable_multiverse).get("steps") |
| if isinstance(steps, bool) or not isinstance(steps, int) or steps < 1: |
| raise ValueError("committed step count is missing") |
| return steps |
|
|
|
|
| def _hnc_doc() -> str: |
| brain = _brain() |
| on = json.dumps(_path_block(brain, True)["coherence"]) |
| off = json.dumps(_path_block(brain, False)["coherence"]) |
| return f"""Call HarmonicRealityField.step. Label: inferred. This is a rerun in this process. |
| |
| The class default HarmonicRealityField() sets enable_multiverse=True (multiverse on). This tool's default is that class default. The output field enable_multiverse is the value used for this call. |
| |
| Do not mix the two committed brain paths. Cited, huggingface/hnc-field/replay/brain.json: multiverse on coherence is {on}; multiverse off coherence is {off}. |
| |
| steps is capped at the committed steps field for the selected path. There is no frequencies argument. Trade readiness: HOLD. {HOLD_LINE} |
| |
| Args: |
| alpha: Observer gain passed to HarmonicRealityField. Label: cited, DEFAULT_ALPHA in aureon/harmonic/aureon_harmonic_reality.py. |
| beta: Echo gain passed to HarmonicRealityField. Label: cited, DEFAULT_BETA. |
| tau: Echo delay passed to HarmonicRealityField. Label: cited, DEFAULT_TAU. |
| g: Observer nonlinearity passed to HarmonicRealityField. Label: cited, DEFAULT_G. |
| delta_t: Observer window passed to HarmonicRealityField. Label: cited, DEFAULT_DELTA_T. |
| sample_rate: Samples per second. Signature default 1000.0. Label: cited, HarmonicRealityField.__init__. |
| enable_multiverse: Class default is True (multiverse on). False selects the other committed brain path. Reported in the output. Label: cited. |
| steps: Step count. Omitted uses the committed steps for the selected path. Values above that cap are unavailable. Label: cited, vendored brain.json steps. |
| """ |
|
|
|
|
| @quiet |
| def hnc_reality_step( |
| alpha: float = 0.85, |
| beta: float = 0.9, |
| tau: float = 0.05, |
| g: float = 1.5, |
| delta_t: float = 0.01, |
| sample_rate: float = 1000.0, |
| enable_multiverse: bool = True, |
| steps: int | None = None, |
| ) -> str: |
| turned_on = enable_multiverse is True |
| if not isinstance(enable_multiverse, bool): |
| return _dumps( |
| { |
| "class_default_enable_multiverse": True, |
| "enable_multiverse": None, |
| "label": "cited", |
| "reason": "enable_multiverse is missing", |
| "status": "unavailable", |
| } |
| ) |
| try: |
| cap = _cap(turned_on) |
| except (OSError, ValueError, KeyError, TypeError): |
| return _dumps( |
| { |
| "class_default_enable_multiverse": True, |
| "enable_multiverse": turned_on, |
| "label": "cited", |
| "reason": "The committed step count is missing.", |
| "status": "unavailable", |
| } |
| ) |
| chosen: int | None |
| if steps is None: |
| chosen = cap |
| elif isinstance(steps, bool) or not isinstance(steps, int) or steps < 1: |
| chosen = None |
| elif steps > cap: |
| return _dumps( |
| { |
| "class_default_enable_multiverse": True, |
| "enable_multiverse": turned_on, |
| "label": "cited. The cap is the steps field on the selected brain path.", |
| "reason": "steps is above the committed replay length for this path", |
| "status": "unavailable", |
| "steps_cap": cap, |
| } |
| ) |
| else: |
| chosen = steps |
| numbers = { |
| "alpha": _finite(alpha), |
| "beta": _finite(beta), |
| "tau": _finite(tau), |
| "g": _finite(g), |
| "delta_t": _finite(delta_t), |
| "sample_rate": _finite(sample_rate), |
| } |
| if ( |
| chosen is None |
| or any(value is None for value in numbers.values()) |
| or numbers["sample_rate"] is None |
| or numbers["sample_rate"] <= 0 |
| ): |
| return _dumps( |
| { |
| "class_default_enable_multiverse": True, |
| "enable_multiverse": turned_on, |
| "label": "cited", |
| "reason": "A parameter is missing.", |
| "status": "unavailable", |
| "steps_cap": cap, |
| } |
| ) |
| from aureon.harmonic.aureon_harmonic_reality import HarmonicRealityField |
|
|
| field = HarmonicRealityField( |
| alpha=float(numbers["alpha"]), |
| beta=float(numbers["beta"]), |
| tau=float(numbers["tau"]), |
| g=float(numbers["g"]), |
| delta_t=float(numbers["delta_t"]), |
| sample_rate=float(numbers["sample_rate"]), |
| enable_multiverse=turned_on, |
| ) |
| series = [float(field.step()) for _ in range(chosen)] |
| payload: dict[str, Any] = { |
| "class_default_enable_multiverse": True, |
| "coherence": float(field.coherence), |
| "committed_replay_match": None, |
| "committed_replay_path": "multiverse_extension" if turned_on else "hnc", |
| "enable_multiverse": bool(field.enable_multiverse), |
| "get_state": field.get_state(), |
| "label": "inferred. Rerun of HarmonicRealityField.step in this process.", |
| "lambda_last": series[-1], |
| "series": series, |
| "state_name": field.state.name, |
| "state_value": field.state.value, |
| "status": "inferred", |
| "steps": chosen, |
| "steps_cap": cap, |
| } |
| payload["committed_replay_match"] = _committed_match( |
| payload, |
| {key: float(value) for key, value in numbers.items()}, |
| turned_on, |
| ) |
| return _dumps(payload) |
|
|
|
|
| def _committed_match(payload: dict[str, Any], numbers: dict[str, float], enable_multiverse: bool) -> bool | None: |
| brain = _brain() |
| parameters = brain.get("hnc", {}).get("parameters") |
| if not isinstance(parameters, dict): |
| return None |
| expected = { |
| "alpha": parameters.get("alpha"), |
| "beta": parameters.get("beta"), |
| "delta_t": parameters.get("delta_t"), |
| "g": parameters.get("g"), |
| "sample_rate": parameters.get("sample_rate"), |
| "tau": parameters.get("tau"), |
| } |
| if any(numbers[key] != expected[key] for key in expected): |
| return None |
| block = _path_block(brain, enable_multiverse) |
| if payload["steps"] != block.get("steps"): |
| return None |
| fresh = { |
| "coherence": payload["coherence"], |
| "lambda_last": payload["lambda_last"], |
| "series": payload["series"], |
| } |
| rounded = json.loads(json.dumps(fresh)) |
| return ( |
| rounded["coherence"] == block.get("coherence") |
| and rounded["lambda_last"] == block.get("lambda_last") |
| and rounded["series"] == block.get("series") |
| and payload["state_value"] == block.get("state") |
| ) |
|
|
|
|
| hnc_reality_step.__doc__ = _hnc_doc() |
|
|
|
|
| @quiet |
| def hnc_canonical_read(envelope_jsonl: str | None = None) -> str: |
| """Read CanonicalField. An empty envelope does not touch state/. |
| |
| The public JSON drops consciousness_level and consciousness_psi. Those fields have defaults on CanonicalField, so CanonicalField(**public_dict) still constructs and does not raise. It is not a full to_dict() round-trip: the rebuilt fields are the defaults, not whatever to_dict() would have carried. |
| |
| Label: inferred when an envelope was supplied and available is true. Otherwise the field is unavailable. Cited: read_canonical_field does not run unless a temp trace was set. The bundled brain replay is a formula record, not a canonical envelope. |
| |
| Args: |
| envelope_jsonl: Optional JSONL text for one canonical envelope. Empty or omitted returns the unavailable field and does not read state/. Label: cited, aureon/core/hnc_field.py read_canonical_field. |
| """ |
|
|
| if envelope_jsonl is None or not str(envelope_jsonl).strip(): |
| from aureon.core.hnc_field import CanonicalField |
|
|
| body = CanonicalField().to_dict() |
| body["label"] = ( |
| "cited. No envelope was supplied, so state/ was not read. " |
| "The bundled brain replay is a formula record, not a canonical envelope." |
| ) |
| body["status"] = "unavailable" |
| return _dumps(_omit_public_keys(body)) |
| return _dumps(_omit_public_keys(_with_envelope(str(envelope_jsonl), blend=False))) |
|
|
|
|
| @quiet |
| def hnc_blend_read(envelope_jsonl: str | None = None) -> str: |
| """Read BlendedField. Does not call publish_subfield. An empty envelope does not touch state/. |
| |
| BlendedField.to_dict does not include consciousness_level or consciousness_psi. The same two keys are dropped if they appear. That drop does not change a BlendedField round-trip. |
| |
| Label: inferred when an envelope was supplied and available is true. Otherwise unavailable. |
| |
| Args: |
| envelope_jsonl: Optional JSONL text. Empty or omitted returns the unavailable blend and does not read state/. Label: cited, aureon/core/hnc_field.py blend_field. |
| """ |
|
|
| if envelope_jsonl is None or not str(envelope_jsonl).strip(): |
| from aureon.core.hnc_field import BlendedField |
|
|
| body = BlendedField().to_dict() |
| body["label"] = "cited. No envelope was supplied, so state/ was not read." |
| body["status"] = "unavailable" |
| return _dumps(_omit_public_keys(body)) |
| return _dumps(_omit_public_keys(_with_envelope(str(envelope_jsonl), blend=True))) |
|
|
|
|
| def _with_envelope(text: str, *, blend: bool) -> dict[str, Any]: |
| from aureon.core.hnc_field import blend_field, read_canonical_field |
|
|
| previous_trace = os.environ.get("AUREON_HNC_TRACE_PATH") |
| previous_bus = os.environ.get("AUREON_BUS_TRACE_DIR") |
| directory = Path(tempfile.mkdtemp(prefix="aureon-public-field-")) |
| trace = directory / "envelope.jsonl" |
| bus_dir = directory / "bus" |
| bus_dir.mkdir() |
| try: |
| payload = text if text.endswith("\n") else text + "\n" |
| trace.write_text(payload, encoding="utf-8") |
| os.environ["AUREON_HNC_TRACE_PATH"] = str(trace) |
| os.environ["AUREON_BUS_TRACE_DIR"] = str(bus_dir) |
| field = read_canonical_field(bus=object()) |
| if blend: |
| body = blend_field(bus=object()).to_dict() |
| else: |
| body = field.to_dict() |
| finally: |
| if previous_trace is None: |
| os.environ.pop("AUREON_HNC_TRACE_PATH", None) |
| else: |
| os.environ["AUREON_HNC_TRACE_PATH"] = previous_trace |
| if previous_bus is None: |
| os.environ.pop("AUREON_BUS_TRACE_DIR", None) |
| else: |
| os.environ["AUREON_BUS_TRACE_DIR"] = previous_bus |
| trace.unlink(missing_ok=True) |
| for child in bus_dir.glob("*"): |
| child.unlink(missing_ok=True) |
| bus_dir.rmdir() |
| directory.rmdir() |
| body["label"] = "inferred. Read from the envelope supplied to this process." |
| body["status"] = "inferred" if body.get("available") is True else "unavailable" |
| return body |
|
|
|
|
| @quiet |
| def lambda_engine_step( |
| volatility: float | None = None, |
| readings: list[dict[str, Any]] | None = None, |
| ) -> str: |
| """Run LambdaEngine.step, or return status unavailable. |
| |
| This is not HarmonicRealityField and it is not the committed brain replay. The committed replay ships no readings. A missing or short reading list returns status unavailable and does not fill lambda or coherence with 0. Auris-to-Lambda link: HOLD. |
| |
| Args: |
| volatility: Finite volatility. Omitted is unavailable, not 0. Label: cited, LambdaEngine.step. |
| readings: Objects with name, value, confidence, and state. Fewer than two returns status unavailable. Label: cited, huggingface/hnc-field/app.py. Auris-to-Lambda link: HOLD. |
| """ |
|
|
| parsed, reason = _parse_readings(readings) |
| vol = _finite(volatility) |
| if parsed is None or vol is None: |
| why = reason or "volatility is missing" |
| return _dumps( |
| { |
| "auris_to_lambda": "HOLD", |
| "label": ( |
| "cited. The committed brain replay ships no LambdaEngine readings. " |
| "This is not HarmonicRealityField. Auris-to-Lambda link: HOLD." |
| ), |
| "lambda_t": None, |
| "reason": why, |
| "status": "unavailable", |
| } |
| ) |
| fd, name = tempfile.mkstemp(prefix="aureon-public-lambda-", suffix=".json") |
| os.close(fd) |
| path = Path(name) |
| try: |
| from aureon.core.aureon_lambda_engine import LambdaEngine, SubsystemReading |
|
|
| engine = LambdaEngine(state_path=path) |
| state = engine.step(parsed_readings(parsed, SubsystemReading), volatility=vol, auto_persist=False) |
| body = state.to_dict() |
| finally: |
| path.unlink(missing_ok=True) |
| body["auris_to_lambda"] = "HOLD" |
| body["label"] = ( |
| "inferred. LambdaEngine.step on the supplied readings. " |
| "Not HarmonicRealityField. Not the committed brain replay. " |
| "Field names are the keys LambdaState.to_dict already returns. " |
| "Auris-to-Lambda link: HOLD." |
| ) |
| body["status"] = "inferred" |
| return _dumps(body) |
|
|
|
|
| def parsed_readings(rows: list[dict[str, Any]], reading_type: Any) -> list[Any]: |
| return [ |
| reading_type( |
| name=row["name"], |
| value=row["value"], |
| confidence=row["confidence"], |
| state=row["state"], |
| ) |
| for row in rows |
| ] |
|
|
|
|
| def _parse_readings(readings: list[dict[str, Any]] | None) -> tuple[list[dict[str, Any]] | None, str | None]: |
| if readings is None: |
| return None, "LambdaEngine path: unavailable. No readings were supplied." |
| if not isinstance(readings, list) or len(readings) < 2: |
| return None, "LambdaEngine path: unavailable. At least two readings are required." |
| parsed: list[dict[str, Any]] = [] |
| for row in readings: |
| if not isinstance(row, dict): |
| return None, "LambdaEngine path: unavailable. A reading is missing fields." |
| name = row.get("name") |
| value = _finite(row.get("value")) |
| confidence = _finite(row.get("confidence")) |
| state = row.get("state") |
| if not isinstance(name, str) or not name.strip() or value is None or confidence is None: |
| return None, "LambdaEngine path: unavailable. A reading is missing fields." |
| if not isinstance(state, str) or not state.strip(): |
| return None, "LambdaEngine path: unavailable. A reading is missing fields." |
| parsed.append( |
| { |
| "name": name.strip(), |
| "value": value, |
| "confidence": confidence, |
| "state": state.strip(), |
| } |
| ) |
| return parsed, None |
|
|
|
|
| def _texture( |
| volatility: float | None, |
| momentum: float | None, |
| volume: float | None, |
| spread: float | None, |
| ) -> dict[str, float] | None: |
| supplied: dict[str, float] = {} |
| for key, raw in ( |
| ("volatility", volatility), |
| ("momentum", momentum), |
| ("volume", volume), |
| ("spread", spread), |
| ): |
| if raw is None: |
| continue |
| number = _finite(raw) |
| if number is None: |
| continue |
| supplied[key] = number |
| return supplied or None |
|
|
|
|
| def _queen() -> Any: |
| from aureon.utils.aureon_queen_hive_mind import QueenHiveMind |
|
|
| return QueenHiveMind.__new__(QueenHiveMind) |
|
|
|
|
| @quiet |
| def auris_nodes( |
| volatility: float | None = None, |
| momentum: float | None = None, |
| volume: float | None = None, |
| spread: float | None = None, |
| ) -> str: |
| """Nine nodes. Real function: QueenHiveMind.read_auris_nodes. |
| |
| This tool is the nine nodes. It is not the conjecture (not called), not the persona (not scored), and not the coherence evaluator. The coherence evaluator is auris_coherence. Only the nine nodes and get_auris_coherence are real functions exposed here. |
| |
| A missing field is omitted. Nothing is filled with zero. Label: inferred. |
| |
| Args: |
| volatility: Optional Tiger and related input. Omitted stays omitted. Label: cited, _auris_input. |
| momentum: Optional Falcon and Dolphin input. Omitted stays omitted. Label: cited, _auris_input. |
| volume: Optional Falcon and CargoShip input. Omitted stays omitted. Label: cited, _auris_input. |
| spread: Optional Deer input. Omitted stays omitted. Label: cited, _auris_input. |
| """ |
|
|
| market = _texture(volatility, momentum, volume, spread) |
| nodes = _queen().read_auris_nodes(market) |
| missing = sorted(name for name, row in nodes.items() if row.get("value") is None) |
| return _dumps( |
| { |
| "auris": "nine nodes", |
| "label": "inferred. QueenHiveMind.read_auris_nodes. A missing field is not filled.", |
| "missing_nodes": missing, |
| "nodes": nodes, |
| "which_auris": AURIS_FOUR, |
| } |
| ) |
|
|
|
|
| @quiet |
| def auris_coherence( |
| volatility: float | None = None, |
| momentum: float | None = None, |
| volume: float | None = None, |
| spread: float | None = None, |
| ) -> str: |
| """Coherence evaluator. Real function: QueenHiveMind.get_auris_coherence. |
| |
| This tool is the coherence evaluator. It is not the conjecture (not called), not the nine nodes (those are read_auris_nodes and AURIS_NODES on auris_nodes), and not the persona (not scored). Only the nine nodes and get_auris_coherence are real functions exposed here. |
| |
| Whenever any of the nine nodes has no reading, status is no_data and coherence is null. The missing node names are listed. Nothing is filled with zero. Auris-to-Lambda link: HOLD. Trade readiness: HOLD. |
| |
| Args: |
| volatility: Optional. Omitted is not filled with zero. Label: cited, _auris_input. |
| momentum: Optional. Omitted is not filled with zero. Label: cited, _auris_input. |
| volume: Optional. Omitted is not filled with zero. Label: cited, _auris_input. |
| spread: Optional. Omitted is not filled with zero. Label: cited, _auris_input. |
| """ |
|
|
| market = _texture(volatility, momentum, volume, spread) |
| queen = _queen() |
| nodes = queen.read_auris_nodes(market) |
| coherence, status = queen.get_auris_coherence(market) |
| missing = sorted(name for name, row in nodes.items() if row.get("value") is None) |
| if missing or coherence is None: |
| return _dumps( |
| { |
| "auris": "coherence evaluator", |
| "auris_to_lambda": "HOLD", |
| "coherence": None, |
| "label": "inferred. get_auris_coherence returned no complete nine-node reading. Auris-to-Lambda link: HOLD.", |
| "missing_nodes": missing, |
| "status": "no_data", |
| "trade_readiness": "HOLD", |
| "which_auris": AURIS_FOUR, |
| } |
| ) |
| return _dumps( |
| { |
| "auris": "coherence evaluator", |
| "auris_to_lambda": "HOLD", |
| "coherence": coherence, |
| "label": "inferred. This is the evaluator return, not a track record. Auris-to-Lambda link: HOLD.", |
| "missing_nodes": missing, |
| "status": status, |
| "trade_readiness": "HOLD", |
| "which_auris": AURIS_FOUR, |
| } |
| ) |
|
|
|
|
| def _lambda_column() -> list[float]: |
| path = paths.qgita_series() |
| values: list[float] = [] |
| with path.open(encoding="utf-8", newline="") as handle: |
| reader = csv.DictReader(handle) |
| if reader.fieldnames is None or "Lambda" not in reader.fieldnames: |
| return [] |
| for row in reader: |
| try: |
| values.append(float(row["Lambda"])) |
| except (TypeError, ValueError): |
| continue |
| return values |
|
|
|
|
| def _event_record(item: Any, kind: str) -> dict[str, Any]: |
| if kind == "ftcp": |
| return { |
| "adapter": "QGITAMarketAnalyzer.analyze", |
| "curvature": float(item.curvature), |
| "is_valid": bool(item.is_valid), |
| "kind": "FTCP", |
| "knot_index": int(item.knot_index), |
| "timestamp": float(item.timestamp), |
| } |
| return { |
| "adapter": "QGITAMarketAnalyzer.analyze", |
| "event_type": str(item.event_type.value), |
| "kind": "lighthouse", |
| "lighthouse_intensity": float(item.lighthouse_intensity), |
| "timestamp": float(item.timestamp), |
| } |
|
|
|
|
| @quiet |
| def qgita_analyze( |
| values: list[float] | None = None, |
| timestamps: list[float] | None = None, |
| ) -> str: |
| """Run QGITAMarketAnalyzer.analyze and drop the signals object. |
| |
| Events are the adapter return, not a detection rate. Detection rate: HOLD. False-alarm rate: HOLD. The bundled series is cited-unverified and is not public market data. |
| |
| Args: |
| values: Optional series. Omitted loads the vendored de440 Lambda column. A supplied series is inferred and is not the committed replay. Length must be at least 10. Label: cited for the file, inferred for a pasted series. |
| timestamps: Optional clocks, one per value. Omitted uses the sample index. Label: cited, the Space clock. |
| """ |
|
|
| supplied = values is not None |
| series = list(values) if supplied else _lambda_column() |
| if supplied: |
| cleaned: list[float] = [] |
| for item in series: |
| number = _finite(item) |
| if number is None: |
| return _dumps( |
| { |
| "detection_rate": "HOLD", |
| "false_alarm_rate": "HOLD", |
| "label": "inferred", |
| "reason": "A value is missing.", |
| "status": "unavailable", |
| } |
| ) |
| cleaned.append(number) |
| series = cleaned |
| if len(series) < 10: |
| return _dumps( |
| { |
| "detection_rate": "HOLD", |
| "false_alarm_rate": "HOLD", |
| "label": "cited. analyze returns insufficient_data below 10 samples.", |
| "reason": "At least ten values are required.", |
| "status": "unavailable", |
| } |
| ) |
| if timestamps is None: |
| clocks = [float(index) for index in range(len(series))] |
| clock_label = "cited. Sample index, the Space clock." |
| else: |
| if len(timestamps) != len(series): |
| return _dumps( |
| { |
| "detection_rate": "HOLD", |
| "false_alarm_rate": "HOLD", |
| "label": "inferred", |
| "reason": "timestamps and values differ in length.", |
| "status": "unavailable", |
| } |
| ) |
| clocks = [] |
| for item in timestamps: |
| number = _finite(item) |
| if number is None: |
| return _dumps( |
| { |
| "detection_rate": "HOLD", |
| "false_alarm_rate": "HOLD", |
| "label": "inferred", |
| "reason": "A timestamp is missing.", |
| "status": "unavailable", |
| } |
| ) |
| clocks.append(number) |
| clock_label = "inferred. Caller-supplied clocks." |
| from aureon.wisdom.aureon_qgita_framework import QGITAMarketAnalyzer |
|
|
| analyzer = QGITAMarketAnalyzer() |
| for index, value in enumerate(series): |
| analyzer.feed_price(float(value), timestamp=clocks[index]) |
| raw = analyzer.analyze() |
| stage1 = raw.get("stage1") if isinstance(raw.get("stage1"), dict) else {} |
| stage2 = raw.get("stage2") if isinstance(raw.get("stage2"), dict) else {} |
| valid = list(stage1.get("valid_ftcps") or []) |
| events = list(stage2.get("lighthouse_events") or []) |
| body: dict[str, Any] = { |
| "clock": clock_label, |
| "coherence": raw.get("coherence"), |
| "detection_rate": "HOLD", |
| "false_alarm_rate": "HOLD", |
| "ftcp_count": stage1.get("ftcp_count"), |
| "ftcps": [_event_record(item, "ftcp") for item in valid], |
| "label": ( |
| "cited-unverified bundled series. Not public market data." |
| if not supplied |
| else "inferred. Adapter run on the supplied series. Not the committed replay." |
| ), |
| "lhe_count": stage2.get("lhe_count"), |
| "lighthouse_events": [_event_record(item, "lhe") for item in events], |
| "regime": raw.get("regime"), |
| "samples_fed": len(series), |
| "status": raw.get("status", "unavailable"), |
| } |
| return _dumps(body) |
|
|
|
|
| @quiet |
| def organism_replay_read() -> str: |
| """Read the committed circulation replay file. This does not start a seat. |
| |
| Connectivity figures are cited from the file. The fault block is seeded synthetic input, not telemetry. A claim that those faults are recorded seat faults is HOLD. |
| |
| Args: |
| none: This tool takes no inputs. Label: cited, huggingface/organism-replay/replay/organism.json. |
| """ |
|
|
| record = _load(paths.organism_json()) |
| connectivity = record.get("connectivity") if isinstance(record.get("connectivity"), dict) else {} |
| self_model = record.get("self_model") if isinstance(record.get("self_model"), dict) else {} |
| prediction = self_model.get("prediction") if isinstance(self_model.get("prediction"), dict) else {} |
| fault = self_model.get("fault") if isinstance(self_model.get("fault"), dict) else {} |
| seats = connectivity.get("seats") if isinstance(connectivity.get("seats"), dict) else {} |
| isolated = connectivity.get("isolated") |
| topics = _dashboard_topics() |
| return _dumps( |
| { |
| "cited": { |
| "connected": connectivity.get("connected"), |
| "connectivity_score": connectivity.get("score"), |
| "http_seat_bound": connectivity.get("http_seat_bound"), |
| "isolated": isolated, |
| "label": "cited, huggingface/organism-replay/replay/organism.json", |
| "prediction_accuracy": prediction.get("accuracy"), |
| "prediction_baseline_accuracy": prediction.get("baseline_accuracy"), |
| "roster": connectivity.get("roster"), |
| "seats": seats, |
| "verdict": connectivity.get("verdict"), |
| }, |
| "dashboard_topics": topics, |
| "label": "cited replay file. Circulation was not started.", |
| "seeded_synthetic_fault": { |
| "detection_rate_in_file": fault.get("detection_rate"), |
| "false_alarm_rate_in_file": fault.get("false_alarm_rate"), |
| "hold": ( |
| "A claim that these are recorded seat faults is HOLD. " |
| "These figures are not a QGITA detection rate and not telemetry." |
| ), |
| "label": "Seeded synthetic input, not telemetry.", |
| "origin": fault.get("origin"), |
| "seed": fault.get("seed"), |
| }, |
| } |
| ) |
|
|
|
|
| def _dashboard_topics() -> list[str]: |
| topics = ( |
| "ics.body.state", |
| "ics.mind.state", |
| "ics.source.state", |
| "ics.soul.state", |
| "ics.hnc.state", |
| "ics.monologue", |
| ) |
| try: |
| from aureon.core.aureon_thought_bus import ThoughtBus |
| from aureon.core.cognitive_dashboard import CognitiveDashboard |
|
|
| bus = ThoughtBus(persist_path=None) |
| CognitiveDashboard(thought_bus=bus) |
| except Exception: |
| return [f"Dashboard topic {topic}: unavailable" for topic in topics] |
| lines = [] |
| for topic in topics: |
| found = bus.recall(topic, limit=1) |
| if found: |
| lines.append(f"Dashboard topic {topic}: present in the replay") |
| else: |
| lines.append(f"Dashboard topic {topic}: unavailable") |
| return lines |
|
|
|
|
| @quiet |
| def organism_catalog() -> str: |
| """List circulation seats, algorithmic nodes, and deterministic systems. |
| |
| These are names in source. This tool does not start circulation, self-repair, or self-code. Label: cited. |
| |
| Args: |
| none: This tool takes no inputs. Label: cited, circulation_seats, list_nodes, and list_systems. |
| """ |
|
|
| from aureon.core.algorithmic_nodes import list_nodes |
| from aureon.core.deterministic_systems import list_systems |
| from aureon.core.organism_circulation import circulation_seats |
|
|
| return _dumps( |
| { |
| "label": "cited. Names in source. Seats are not running.", |
| "nodes": list_nodes(), |
| "seats": list(circulation_seats()), |
| "systems": list_systems(), |
| } |
| ) |
|
|
