224 lines
8.3 KiB
Python
224 lines
8.3 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import argparse
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
from typing import Any
|
|
|
|
from .schemas import FinancialReasoningRequest, FinancialReasoningResponse
|
|
|
|
|
|
TOKEN_PATTERN = re.compile(r"[A-Za-z0-9_]{2,}")
|
|
|
|
|
|
class InvalidMemoryRecord(ValueError):
|
|
"""Raised when a record cannot safely become retrieval memory."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class StoredMemoryRecord:
|
|
content_hash: str
|
|
payload: dict[str, Any]
|
|
|
|
|
|
class BlumFinanceMemoryStore:
|
|
"""Small, auditable local memory with strict point-in-time retrieval.
|
|
|
|
This store never changes model weights. It exposes only matured observations
|
|
available before a new request and labels them as historical analogies.
|
|
"""
|
|
|
|
def __init__(self, path: str | Path) -> None:
|
|
self.path = Path(path).expanduser().resolve()
|
|
|
|
def add(self, payload: dict[str, Any]) -> StoredMemoryRecord:
|
|
normalized = _validate_memory_payload(payload)
|
|
content_hash = _content_hash(normalized)
|
|
row = {"content_hash": content_hash, "payload": normalized}
|
|
existing = self._rows()
|
|
if not any(item.get("content_hash") == content_hash for item in existing):
|
|
self._replace([*existing, row])
|
|
return StoredMemoryRecord(content_hash=content_hash, payload=normalized)
|
|
|
|
def add_bundle(self, bundle: str | Path | dict[str, Any]) -> StoredMemoryRecord:
|
|
from .contributions import validate_contribution_bundle
|
|
|
|
if isinstance(bundle, (str, Path)):
|
|
value = json.loads(Path(bundle).read_text(encoding="utf-8"))
|
|
else:
|
|
value = bundle
|
|
validation = validate_contribution_bundle(value)
|
|
if not validation.accepted:
|
|
raise InvalidMemoryRecord(
|
|
"Contribution is not eligible for memory: " + ", ".join(validation.blockers)
|
|
)
|
|
return self.add(value["payload"])
|
|
|
|
def retrieve(
|
|
self,
|
|
request: FinancialReasoningRequest,
|
|
*,
|
|
limit: int = 3,
|
|
) -> list[dict[str, Any]]:
|
|
if limit <= 0:
|
|
return []
|
|
request_tokens = _request_tokens(request.model_dump(mode="json"))
|
|
candidates: list[tuple[float, datetime, dict[str, Any]]] = []
|
|
for row in self._rows():
|
|
payload = row.get("payload")
|
|
if not isinstance(payload, dict):
|
|
continue
|
|
try:
|
|
normalized = _validate_memory_payload(payload)
|
|
observed_at = _timestamp(normalized["outcome"]["observed_at"])
|
|
except (InvalidMemoryRecord, KeyError, TypeError, ValueError):
|
|
continue
|
|
if observed_at > _aware(request.as_of):
|
|
continue
|
|
memory_request = normalized["request"]
|
|
memory_tokens = _request_tokens(memory_request)
|
|
overlap = len(request_tokens & memory_tokens) / max(1, len(request_tokens | memory_tokens))
|
|
ticker_match = str(memory_request.get("ticker", "")).upper() == request.ticker.upper()
|
|
horizon_match = str(memory_request.get("horizon", "")) == request.horizon
|
|
score = overlap + (2.0 if ticker_match else 0.0) + (0.5 if horizon_match else 0.0)
|
|
candidates.append((score, observed_at, _retrieval_payload(normalized, row.get("content_hash"))))
|
|
candidates.sort(key=lambda item: (item[0], item[1]), reverse=True)
|
|
return [item[2] for item in candidates[:limit] if item[0] > 0]
|
|
|
|
def _rows(self) -> list[dict[str, Any]]:
|
|
if not self.path.is_file():
|
|
return []
|
|
rows: list[dict[str, Any]] = []
|
|
for line in self.path.read_text(encoding="utf-8").splitlines():
|
|
try:
|
|
value = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if isinstance(value, dict):
|
|
rows.append(value)
|
|
return rows
|
|
|
|
def _replace(self, rows: list[dict[str, Any]]) -> None:
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = self.path.with_suffix(self.path.suffix + ".tmp")
|
|
body = "".join(_canonical_json(row) + "\n" for row in rows)
|
|
temporary.write_text(body, encoding="utf-8")
|
|
os.replace(temporary, self.path)
|
|
|
|
|
|
def _validate_memory_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
|
if not isinstance(payload, dict):
|
|
raise InvalidMemoryRecord("Memory payload must be an object")
|
|
request = payload.get("request")
|
|
response = payload.get("response")
|
|
outcome = payload.get("outcome")
|
|
quality = payload.get("quality")
|
|
try:
|
|
parsed_request = FinancialReasoningRequest.model_validate(request)
|
|
parsed_response = FinancialReasoningResponse.model_validate(response)
|
|
except Exception as exc:
|
|
raise InvalidMemoryRecord(f"Invalid BLUM request or response: {exc}") from exc
|
|
if not isinstance(outcome, dict) or not outcome.get("observed_at"):
|
|
raise InvalidMemoryRecord("A matured outcome with observed_at is required")
|
|
observed_at = _timestamp(outcome["observed_at"])
|
|
if observed_at <= _aware(parsed_request.as_of):
|
|
raise InvalidMemoryRecord("The outcome must be observed after the decision")
|
|
status = str(outcome.get("status") or "").strip().lower()
|
|
if status in {"", "pending", "unresolved", "inconclusive"}:
|
|
raise InvalidMemoryRecord("The outcome is not mature")
|
|
if not isinstance(quality, dict) or quality.get("source_verified") is not True:
|
|
raise InvalidMemoryRecord("Memory requires verified source provenance")
|
|
try:
|
|
score = float(quality.get("score"))
|
|
except (TypeError, ValueError) as exc:
|
|
raise InvalidMemoryRecord("Memory quality score is missing") from exc
|
|
if score < 70:
|
|
raise InvalidMemoryRecord("Memory quality is below 70")
|
|
return {
|
|
**payload,
|
|
"request": parsed_request.model_dump(mode="json"),
|
|
"response": parsed_response.model_dump(mode="json"),
|
|
"outcome": dict(outcome),
|
|
"quality": {**quality, "score": score},
|
|
}
|
|
|
|
|
|
def _retrieval_payload(payload: dict[str, Any], content_hash: Any) -> dict[str, Any]:
|
|
request = payload["request"]
|
|
response = payload["response"]
|
|
outcome = payload["outcome"]
|
|
return {
|
|
"memory_id": str(content_hash or _content_hash(payload)),
|
|
"ticker": request.get("ticker"),
|
|
"horizon": request.get("horizon"),
|
|
"decision_as_of": request.get("as_of"),
|
|
"observed_at": outcome.get("observed_at"),
|
|
"prior_status": response.get("status"),
|
|
"prior_thesis": response.get("thesis"),
|
|
"outcome": {
|
|
key: outcome.get(key)
|
|
for key in ("status", "realized_r", "benchmark_excess")
|
|
if outcome.get(key) is not None
|
|
},
|
|
"lesson": outcome.get("lesson") or "No explicit lesson was supplied.",
|
|
"quality_score": payload["quality"]["score"],
|
|
}
|
|
|
|
|
|
def _request_tokens(payload: dict[str, Any]) -> set[str]:
|
|
return {
|
|
token.lower()
|
|
for token in TOKEN_PATTERN.findall(json.dumps(payload, ensure_ascii=False, sort_keys=True))
|
|
}
|
|
|
|
|
|
def _content_hash(payload: dict[str, Any]) -> str:
|
|
return hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _canonical_json(payload: Any) -> str:
|
|
return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
|
|
|
|
def _timestamp(value: Any) -> datetime:
|
|
if isinstance(value, datetime):
|
|
return _aware(value)
|
|
parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
|
|
return _aware(parsed)
|
|
|
|
|
|
def _aware(value: datetime) -> datetime:
|
|
if value.tzinfo is None:
|
|
return value.replace(tzinfo=timezone.utc)
|
|
return value.astimezone(timezone.utc)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(
|
|
description="Import a validated BLUM contribution into local point-in-time memory."
|
|
)
|
|
parser.add_argument("bundle", type=Path)
|
|
parser.add_argument(
|
|
"--memory",
|
|
type=Path,
|
|
default=Path.home() / ".blum-finance" / "memory.jsonl",
|
|
)
|
|
args = parser.parse_args()
|
|
stored = BlumFinanceMemoryStore(args.memory).add_bundle(args.bundle)
|
|
print(
|
|
json.dumps(
|
|
{"status": "stored", "content_hash": stored.content_hash, "memory": str(args.memory)},
|
|
indent=2,
|
|
sort_keys=True,
|
|
)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|