analogos 0.3.0

Programmable analogy framework — five universal primitives for complex systems

analogos

Programmable Analogy Framework · v0.2.0 · GPL-3.0

"Analogy is not a way to explain systems. Analogy is the system."

---

What is analogos?

analogos is a general-purpose framework that formalizes analogy as a programmable primitive.

It is not a machine learning library.
It is not a visualization tool.
It is not a set of didactic metaphors.

It is a systems algebra — a minimal set of operators that describes the behavior of any system where information is generated, distributed, filtered, propagated, and composed.

The central proposition:

Different complex systems — neural networks, financial markets, immune systems, blockchains — are instances of the same structural pattern.

Once that pattern is formalized, any domain becomes a parameterization of the same five primitives. The code is the same. Only the parameters change.

---

Install

pip install analogos

Requires Python 3.10+ and numpy. No other dependencies.

---

The Five Universal Primitives

scan → broadcast → candidate → propagate → compose

Primitive	Role	Complexity
scan()	Traverses the entity space and builds a reference map	O(n) — paid once
broadcast()	Source emits a signal; intensity decays with distance	O(k)
candidate()	Filters entities that received sufficient signal	O(k)
propagate()	Candidates relay signal to neighbors — emergent clusters	O(k·r)
compose()	Aggregates the cluster into a unified output	O(k)

---

Quick Start

Using the Pipeline

import numpy as np
from analogos import Entity, Pipeline, PipelineConfig

# define your entity space
tokens = [
    Entity('t1', 'learning',   np.array([0.9, 0.8, 0.1, 0.2]), tags=['concept']),
    Entity('t2', 'neural net', np.array([0.8, 0.7, 0.2, 0.3]), tags=['concept']),
    Entity('t3', 'gradient',   np.array([0.7, 0.5, 0.3, 0.4]), tags=['concept']),
    Entity('t4', 'banana',     np.array([0.1, 0.1, 0.9, 0.8]), tags=['food']),
]

query = Entity('q', 'what is learning?', np.array([0.85, 0.75, 0.15, 0.25]))

pipeline = Pipeline(config=PipelineConfig(
    threshold=0.4,
    falloff='sqrt',      # 'sqrt' | 'linear' | 'quadratic'
    social_factor=0.5,
    mode='top_k',        # 'top_k' | 'weighted_sum' | 'vote' | 'concat'
    top_k=3,
))

result = pipeline.run(source=query, entities=tokens)
print([e.value for e, _ in result.output['result']])
# → ['learning', 'neural net', 'gradient']

Using the primitives directly

from analogos import scan, broadcast, candidate, propagate, compose

ref_map      = scan(tokens)
signals_bc   = broadcast(query, ref_map, intensity=1.0, falloff='sqrt')
candidates   = candidate(tokens, signals_bc, threshold=0.4)
signals_prop = propagate(candidates, tokens, signals_bc, social_factor=0.5)
output       = compose(candidates, signals_prop, mode='top_k', top_k=3)

---

Adaptive Pipeline — Analogical Memory Loop

analogos ships with an adaptive engine that processes text, stores structural patterns in memory, and self-calibrates with each new document.

text → ingest → pipeline → correlate(memory) → adapt config → store → next cycle

from analogos import AdaptivePipeline

ap = AdaptivePipeline()

# cycle 1 — no prior memory
r1 = ap.process("Neural networks learn by adjusting synaptic weights...", doc_id="neuro")
print(r1.summary())

# cycle 2 — correlates with cycle 1, adapts parameters
r2 = ap.process("Financial markets propagate price signals across assets...", doc_id="market")
print(r2.summary())
# → finds shared patterns: ['signals', 'propagation', 'activation', 'patterns']
# → adjusts threshold and social_factor based on structural correlation

# retrieve documents similar to a query
similar = ap.memory.retrieve_similar(source_entity, top_k=3)

Each cycle:

1. Converts raw text into Entity objects via hash-stable random projections (numpy-only, no ML dependencies)
2. Runs the full analogy pipeline
3. Computes Jaccard + cosine similarity against memory
4. Adjusts threshold and social_factor based on cluster density and correlation strength
5. Stores the result for future cycles

---

Domain Instances

Each domain is a parameterization of the five primitives — no new code, only different parameters.

Domain	scan	broadcast	candidate	propagate	compose
analog-attention	tokens	query vector	keys ≥ threshold	value context	weighted sum
analog-neuro	neurons	action potential	threshold neurons	synaptic relay	firing pattern
analog-market	assets	price signal	eligible assets	market contagion	portfolio
analog-immune	antigens	receptor signal	activated cells	immune cascade	response
analog-blockchain	nodes	broadcast tx	validators	peer relay	block commit

analog-neuro

from analogos.domains.neuro import NeuroPipeline, Neuron, Stimulus

neurons = [
    Neuron("LGN_1", "LGN", layer=4, neurotransmitter="glutamate"),
    Neuron("V1_L4", "V1",  layer=4, neurotransmitter="glutamate"),
    Neuron("V2_1",  "V2",  layer=2, neurotransmitter="glutamate"),
    Neuron("MT_1",  "MT",  layer=4, neurotransmitter="glutamate"),
    Neuron("PFC_1", "PFC", layer=3, neurotransmitter="dopamine"),
]
stimulus = Stimulus("s1", "flash visual", modality="visual", target_region="V1")

result = NeuroPipeline().fire(stimulus, neurons)
print(result.summary())
# Caminho de ativação: LGN → V1 → V2 → MT → PFC

analog-market

from analogos.domains.market import MarketPipeline, Asset, MarketEvent

assets = [
    Asset("TLT",  "iShares 20Y Treasury", sector="bonds",      beta=-0.3),
    Asset("XLF",  "Financial Select",     sector="financials", beta=1.3),
    Asset("QQQ",  "NASDAQ-100",           sector="tech",       beta=1.5),
    Asset("GLD",  "Gold Trust",           sector="commodities",beta=-0.1),
]
event = MarketEvent("fed_hike", "Fed rate hike +75bps", magnitude=0.85,
                    affected_sectors=["bonds", "financials"])

result = MarketPipeline().shock(event, assets)
print(result.summary())
# Contágio setorial: bonds → financials → tech → commodities

analog-immune

from analogos.domains.immune import ImmunePipeline, ImmuneCell, Pathogen

cells = [
    ImmuneCell("dc1",    "dendritic",   specificity=0.95),
    ImmuneCell("th1",    "T_helper",    specificity=0.85),
    ImmuneCell("bc_mem", "B_cell",      specificity=0.92, is_memory=True),
    ImmuneCell("tc1",    "T_cytotoxic", specificity=0.90),
]
pathogen = Pathogen("flu", "Influenza H3N2", pathogen_type="virus", virulence=0.75)

result = ImmunePipeline().respond(pathogen, cells)
print(result.summary())
# Resposta: mista (inata + adaptativa) · células de memória: [bc_mem]

analog-blockchain

from analogos.domains.blockchain import BlockchainPipeline, Node, Transaction

nodes = [
    Node("val_01", "validator", stake=0.95, reputation=0.98, region="north_america"),
    Node("val_02", "validator", stake=0.90, reputation=0.96, region="europe"),
    Node("full_01","full_node", stake=0.20, reputation=0.88, region="asia_pacific"),
]
tx = Transaction("tx_001", "DeFi swap 50 ETH → USDC", fee=0.08, priority=0.9)

result = BlockchainPipeline().broadcast_tx(tx, nodes)
print(result.summary())
# Consenso: ✓ ATINGIDO · Bloco: ✓ COMMITADO

---

Project Structure

analogos/                    ← installable package
├── __init__.py              ← full public API
├── py.typed                 ← PEP 561 — IDE autocomplete & type checking
├── core/
│   ├── entity.py            ← Entity base class
│   └── primitives/
│       ├── scan.py
│       ├── broadcast.py
│       ├── candidate.py
│       ├── propagate.py
│       └── compose.py
├── pipeline.py              ← Pipeline · PipelineConfig · PipelineResult
├── ingest.py                ← TextIngester (text → Entity[])
├── memory.py                ← AnalogMemory (incremental record store)
├── correlator.py            ← cross-document structural correlation
└── adaptive.py              ← AdaptivePipeline

examples/                    ← runnable demos
tests/                       ← pytest suite
pyproject.toml               ← pip-installable, PEP 517/518

---

IDE Support

The package ships with py.typed (PEP 561) and full type annotations.
Autocomplete, go-to-definition, and inline docs work out of the box in VS Code, PyCharm, Neovim, and any LSP-compatible editor.

from analogos import Entity, Pipeline, AdaptivePipeline  # fully typed

---

Roadmap

• v0.1.0 — Foundation: five primitives in pure Python + architecture
• v0.2.0 — Installable library: pip install analogos, typed API, adaptive pipeline, memory loop, 11 unit tests
• v0.3.0 — Domains: analog_neuro, analog_market, analog_immune, analog_blockchain — 33 unit tests
• v0.4.0 — Integrations: PyTorch tensors, JAX, benchmark vs dot-product attention
• v1.0.0 — Multi-language bindings (Rust/PyO3, gRPC) + technical paper

---

Theoretical Foundation

Conventional use of analogy in computing is didactic: it explains a hard concept using a familiar domain, then discards the analogy.

analogos inverts this flow.

The central claim is that the structure of the analogy is the structure of the system. The immune system is not "similar to" an attention mechanism. Both are instances of the same formal pattern:

scan → broadcast → candidate → propagate → compose

What does not exist in the literature is the formalization of analogy as a reusable programmable primitive — an operator that can be instantiated across any domain without rewriting the core logic.

That is what analogos proposes.

---

License

GPL-3.0 — use, modify, and redistribute freely. Modified versions must maintain the same license. Source must remain open.

---

Author

Zaqueu Ribeiro · github.com/omega-Core-Dev

---

"The pattern was always there. It just needed a name."