crypto tar

[siriusspinner-stack]

"""
Out of the tar pit — Python shadow (essential core only)
Avoid state. Avoid control. Separate essential / accidental.

This is NOT a full FRP engine.
Just the simplest possible expression of the idea:
pure logic + relations + declarative derivations
"""

from dataclasses import dataclass
from typing import Dict, Set, Tuple, Callable, Any, FrozenSet
from functools import reduce
import operator
from collections import defaultdict

─── tiny immutable relation (set of frozen dicts) ───────────────────────────────

Relation = FrozenSet[Dict[str, Any]]

def project(rel: Relation, *keep_keys: str) -> Relation:
""" π — keep only these attributes """
return frozenset({frozenset({k: r[k] for k in keep_keys if k in r})} for r in rel)

def project_away(rel: Relation, *drop_keys: str) -> Relation:
""" remove these attributes """
return frozenset({frozenset({k: v for k, v in r.items() if k not in drop_keys})} for r in rel)

def select(rel: Relation, pred: Callable[[Dict[str, Any]], bool]) -> Relation:
""" σ — restrict """
return frozenset(r for r in rel if pred(r))

def join(left: Relation, right: Relation) -> Relation:
""" ⋈ natural join on common keys """
common = set(left and left and next(iter(left)).keys()) & set(next(iter(right)).keys())
if not common:
return cartesian(left, right)
result = []
for l in left:
for r in right:
if all(l.get(k) == r.get(k) for k in common):
result.append({**l, **r})
return frozenset(result)

def cartesian(left: Relation, right: Relation) -> Relation:
return frozenset({**l, **r} for l in left for r in right)

def union(left: Relation, right: Relation) -> Relation:
return left | right

def difference(left: Relation, right: Relation) -> Relation:
return left - right

def rename(rel: Relation, mapping: Dict[str, str]) -> Relation:
return frozenset({mapping.get(k, k): v for k, v in r.items()} for r in rel)

─── essential state ─────────────────────────────────────────────────────────────

@DataClass(frozen=True)
class Property:
address: str
price: float
agent: str
date_registered: str

@DataClass(frozen=True)
class Offer:
address: str
offer_price: float
offer_date: str
bidder_name: str
bidder_address: str

@DataClass(frozen=True)
class Decision:
address: str
offer_date: str
bidder_name: str
bidder_address: str
decision_date: str
accepted: bool

example tiny databases (in real system these would come from feeders)

properties = frozenset([
Property("123 Maple", 450000, "alice", "2025-01-10"),
Property("456 Oak", 320000, "bob", "2025-02-01"),
])

offers = frozenset([
Offer("123 Maple", 440000, "2025-03-05", "carol", "789 Pine"),
Offer("123 Maple", 460000, "2025-03-10", "carol", "789 Pine"),
Offer("456 Oak", 310000, "2025-03-08", "dave", "101 Elm"),
])

decisions = frozenset([
Decision("123 Maple", "2025-03-10", "carol", "789 Pine", "2025-03-12", True),
])

─── essential logic ─────────────────────────────────────────────────────────────

def latest_offers(offers: Relation) -> Relation:
""" group by (address, bidder_name, bidder_address) → most recent offer """
groups = defaultdict(list)
for o in offers:
key = (o["address"], o["bidder_name"], o["bidder_address"])
groups[key].append(o)
latest = []
for grp in groups.values():
latest.append(max(grp, key=lambda x: x["offer_date"]))
return frozenset(latest)

def accepted_sales(
decisions: Relation,
current_offers: Relation,
properties: Relation
) -> Relation:
accepted = select(decisions, lambda r: r["accepted"])
joined = join(accepted, current_offers)
return join(joined, frozenset(p.dict for p in properties))

def unsold_properties(properties: Relation, sold: Relation) -> Relation:
prop_keys = frozenset({frozenset({"address": p["address"]}) for p in properties})
sold_keys = frozenset({frozenset({"address": s["address"]}) for s in sold})
return prop_keys - sold_keys

compute views (pure — no side effects)

current_offers_view = latest_offers(frozenset(o.dict for o in offers))
sold_view = accepted_sales(
frozenset(d.dict for d in decisions),
current_offers_view,
frozenset(p.dict for p in properties)
)

print("Current open offers:")
for row in current_offers_view - project(sold_view, "address", "offer_date", "bidder_name", "bidder_address"):
print(row)

print("\nSold:")
for row in sold_view:
print(row)

print("\nStill for sale:")
for row in unsold_properties(frozenset(p.dict for p in properties), sold_view):
print(row)