fraq — Fractal Query Data Library

Model data as infinite, self-similar fractal structures in hyperspace. Each zoom level reveals procedurally generated detail — data exists only virtually and materialises on demand via lazy evaluation.

Core Idea

Imagine a schema that is itself an infinite fractal. You can zoom into any region and receive an unbounded amount of deterministic data in any format you choose (JSON, CSV, YAML, binary). The data doesn't live in memory or on disk — it is generated on the fly from the fractal coordinates and a deterministic seed.

root (depth 0)
 ├── zoom (1,0,0) → child A (depth 1)
 │    ├── zoom (1,0,0) → grandchild AA (depth 2) …
 │    └── zoom (0,1,0) → grandchild AB (depth 2) …
 └── zoom (0,0,1) → child B (depth 1)
      └── … infinite …

Key Properties

Property	Description
Deterministic	Same path → same data, always
Lazy	Nodes materialise only when accessed
Unbounded	No depth limit — zoom forever
Format-agnostic	JSON, CSV, YAML, binary, msgpack-lite, or register your own
Source-agnostic	File, HTTP, SQL, Sensor, Hybrid — same query interface
Schema-exportable	NLP2CMD, OpenAPI, GraphQL, AsyncAPI, gRPC, JSON Schema
Zero storage	Data exists only as computation

Quick Start

pip install -e ".[dev]"

# CLI
fraq explore --dims 3 --depth 5 --format json
fraq stream  --dims 2 --count 20 --format csv
fraq schema  --fields "name:str,value:float,flag:bool" --depth 2
fraq nl "show 10 pdf files created recently" --path .

# Docker
docker compose run test          # run full test suite
docker compose run cli explore --depth 5

Python API

🆕 NEW: Simplified API (v0.2.11+)

The easiest way to use fraq — just generate() data with field specifications:

from fraq import generate, stream, quick_schema, FraqSchema

# 1. generate() — simplest data generation
records = generate({
    'temperature': 'float:10-40',  # 10-40°C range
    'humidity': 'float:0-100',     # 0-100% range
    'sensor_id': 'str',            # auto-formatted IDs
}, count=100)
# Returns: [{'temperature': 23.5, 'humidity': 67.2, 'sensor_id': 'SEN-004572'}, ...]

# 2. stream() — lazy infinite streaming
for record in stream({'cpu': 'float:0-100'}, count=1000, interval=0.1):
    print(record)  # Process one by one

# 3. quick_schema() — auto-detect types with smart transforms
schema = quick_schema('temp', 'humidity', 'pressure')  # auto-ranges for common fields
records = list(schema.records(count=50))

# 4. FraqSchema() without root — auto-created
schema = FraqSchema()  # No need for root=FraqNode(...)
schema.add_field('value', 'float')
records = list(schema.records(count=10))  # Use count instead of depth

See examples/ for complete usage patterns with the new API.

🆕 DataFrame Export (v0.2.11+)

Export directly to Polars, Pandas, or PyArrow — no manual conversion needed:

from fraq import generate

# Polars DataFrame (fast, modern)
df = generate({
    'temperature': 'float:10-40',
    'sensor_id': 'str',
}, count=10000, output='polars')
# Returns: pl.DataFrame

# Pandas DataFrame (familiar API)
df = generate({
    'name': 'str',
    'value': 'float:0-100',
}, count=5000, output='pandas')
# Returns: pd.DataFrame

# PyArrow Table (columnar, zero-copy)
table = generate({
    'timestamp': 'str',
    'reading': 'float',
}, count=1000, output='arrow')
# Returns: pa.Table

# Lazy iterator for streaming
records = generate({'value': 'float'}, count=1000, output='records')
for r in records:  # Memory-efficient streaming
    process(r)

Install extras: pip install fraq[polars] or fraq[pandas] or fraq[arrow]

🆕 IFS Fractal Generator (v0.2.11+)

True structural self-similarity — competitors (Faker, Mimesis, SDV) cannot replicate this:

from fraq import IFSGenerator, AffineTransform, create_ifs

# Organizational hierarchy with fractal structure
ifs = create_ifs("organizational", seed=42)
org_data = ifs.generate(count=1000, depth=3)
# Each level statistically similar to parent

# Custom IFS with transforms
transforms = [
    AffineTransform(scale=0.5, translation=(0.0, 0.0)),    # Department
    AffineTransform(scale=0.3, translation=(0.5, 0.0)),    # Team
    AffineTransform(scale=0.2, translation=(0.8, 0.0)),     # Individual
]
ifs = IFSGenerator(transforms, weights=[0.4, 0.35, 0.25])

# Generate hierarchical data
tree = ifs.generate_hierarchy(
    root={'name': 'HQ', 'budget': 1000000},
    branching=[3, 5, 2],  # 3 depts → 5 teams each → 2 people
    depth=3
)

Use cases:

Testing recursive algorithms with guaranteed input structure
Hierarchical org data with natural distribution
Network topology simulations with repeating patterns

🆕 Faker Integration (v0.2.11+)

Realistic data (names, addresses, PESEL, NIP) in fractal structures:

from fraq import generate

# Mix native fraq types with Faker providers
records = generate({
    'name': 'faker:name',              # Realistic names
    'email': 'faker:email',            # Realistic emails
    'address': 'faker:pl_PL.address', # Polish addresses
    'nip': 'faker:pl_PL.nip',          # Polish NIP
    'pesel': 'faker:pl_PL.pesel',      # Polish PESEL
    'temperature': 'float:10-40',      # Native fraq
}, count=1000)

# Locale-specific generation
records = generate({
    'name': 'faker:pl_PL.name',        # Polish: Jan Kowalski
    'company': 'faker:pl_PL.company',
}, count=100)

Install: pip install fraq[faker]

🆕 pytest Integration (v0.2.11+)

Deterministic test fixtures with fraq:

# conftest.py
import pytest
from fraq.testing import fraq_fixture, fixture_factory

# Method 1: Using fraq_fixture
@pytest.fixture
def sensor_data():
    return fraq_fixture({
        'temperature': 'float:10-40',
        'humidity': 'float:0-100',
        'sensor_id': 'str',
    }, count=100, seed=42)  # Always reproducible!

# Method 2: Using decorator
@fixture_factory({
    'user_id': 'str',
    'age': 'int:18-70',
}, count=50, seed=42)
def users():
    pass  # Fixture auto-generated

# Method 3: Auto-discovered fixtures (no conftest.py needed!)
# pytest automatically provides: fraq_session, fraq_data, fraq_schema

def test_temperature_range(sensor_data):
    assert all(10 <= r['temperature'] <= 40 for r in sensor_data)
    assert len(sensor_data) == 100

def test_with_dataframe(fraq_data):
    df = fraq_data({
        'value': 'float:0-100',
    }, count=1000, output='polars')
    assert df.shape == (1000, 1)

Install: pip install fraq[pytest] Auto-discovery: pytest --fixtures | grep fraq

Classic API

from fraq import FraqNode, FraqSchema, FraqCursor, FormatRegistry, query

# 1. One-liner query
data = query(depth=2, fields=["temp:float", "id:str"], format="csv", limit=10)

# 2. Fluent query builder
from fraq import FraqQuery, FraqExecutor

q = (
    FraqQuery()
    .zoom(5, direction=(0.1, 0.2, 0.7))
    .select("temperature:float", "sensor_id:str", "active:bool")
    .where("temperature", "gt", 0.5)
    .output("json")
    .take(20)
)
result = FraqExecutor(dims=3).execute(q)

# 3. Source adapters — same query, different sources
from fraq import FileAdapter, SQLAdapter, SensorAdapter, HybridAdapter, FileSearchAdapter

# Disk
adapter = FileAdapter()
root = adapter.load_root("gradient_root.json")

# SQL
adapter = SQLAdapter(table="sensors")
root = adapter.load_root("", rows=[{"id": 1, "x": 0.0, "y": 0.0, "z": 0.0}])

# Sensors (infinite stream, zero storage)
adapter = SensorAdapter(base_temp=23.5)
for reading in adapter.stream(depth=3, count=100):
    print(reading)  # {'temperature': 24.1, 'humidity': 58.3, ...}

# Hybrid (merge multiple sources)
hybrid = HybridAdapter()
hybrid.add(FileAdapter(), "local_backup.json")
hybrid.add(SensorAdapter(), "")
merged = hybrid.load_root()

# Files
files = FileSearchAdapter(base_path=".").search(extension="py", limit=5, sort_by="mtime")

# 4. Async streaming (FastAPI SSE, Kafka, NATS)
from fraq.streaming import async_stream
async for record in async_stream(count=1000, interval=0.1):
    yield f"data: {json.dumps(record)}\n\n"

NLP2CMD Integration

New in v0.2.10 — Multi-Model Router & Session Memory

from fraq.text2fraq import ModelRouter, FraqSession

# Route queries to optimal model based on complexity
router = ModelRouter()
model = router.route("find pdf files")          # → 0.5b (fast)
model = router.route("generate complex schema")   # → 7b (accurate)

# Multi-turn conversations with context
session = FraqSession()
session.ask("find 10 pdf files")
session.ask("show as csv")        # follow-up: format changed
session.ask("add 20 more")          # follow-up: limit increased

See examples/v028/ for detailed examples.

NLP2CMD Integration

Export FraqSchema to NLP2CMD's SchemaRegistry for natural language → command transformation:

from fraq import FraqNode, FraqSchema, to_nlp2cmd_schema, to_nlp2cmd_actions

schema = FraqSchema(root=FraqNode(position=(0.0, 0.0, 0.0)))
schema.add_field("temperature", "float")
schema.add_field("sensor_id", "str")

# Command schema → command_schemas/fraq_sensor.json
nlp2cmd_schema = to_nlp2cmd_schema(schema, command_name="fraq_sensor")

# Action registry entries
actions = to_nlp2cmd_actions(schema)
# → fraq_zoom, fraq_query, fraq_stream, fraq_save

# NLP2CMD can now transform:
#   "Show active sensors" → fraq query --fields "temperature:float,sensor_id:str" --depth 2

Schema Export Formats

from fraq import to_openapi, to_graphql, to_asyncapi, to_proto, to_json_schema

to_openapi(schema)     # OpenAPI 3.0 — /zoom, /query, /stream endpoints
to_graphql(schema)     # GraphQL type + Query definitions
to_asyncapi(schema)    # AsyncAPI 3.0 — Kafka/WebSocket channels
to_proto(schema)       # gRPC .proto with FraqService (Zoom + Stream RPCs)
to_json_schema(schema) # JSON Schema for validation

Generators

Generator	Output	Use Case
`HashGenerator`	`float` in configurable range	General-purpose pseudo-random
`FibonacciGenerator`	`float` based on depth	Mathematical sequences
`PerlinGenerator`	Smooth `float` noise	Organic sensor streams
`SensorStreamGenerator`	`dict` with IoT fields	Embedded / edge simulation

Data Source Adapters

Adapter	Source	Key Feature
`FileAdapter`	JSON/YAML/CSV on disk	Save/load roundtrip
`HTTPAdapter`	REST APIs	Fallback to deterministic roots
`SQLAdapter`	PostgreSQL/SQLite	Row mapping, SQL function generation
`SensorAdapter`	IoT streams (RPi/ESP32)	Infinite deterministic readings
`HybridAdapter`	Multiple sources merged	Mean position, XOR seeds
`FileSearchAdapter`	Local filesystem	Fractal file coordinates
`NetworkAdapter`	LAN scanning	Async device/port discovery
`WebCrawlerAdapter`	Website crawling	Async page extraction

Architecture

fraq uses clean architecture principles with separated concerns:

Code Organization

fraq/
├── core.py          # Fractal engine: FraqNode, FraqSchema, FraqCursor
├── api.py           # High-level API: generate(), stream(), quick_schema()
├── formats/         # Serialization (cycle-free structure)
│   ├── registry.py  # FormatRegistry
│   ├── prepare.py   # Data preparation (shared)
│   ├── text.py      # JSON, CSV, YAML
│   └── binary.py    # Binary, msgpack_lite
├── adapters/        # Data source adapters
│   ├── base.py      # BaseAdapter
│   └── file_search.py  # Port/Adapter pattern example
└── cli.py           # Command dispatcher (refactored subparsers)

Design Patterns

Pattern	Where	Benefit
Port/Adapter	`FileSearchAdapter`	I/O separated from business logic, 80%+ pure functions
Pipeline	`generate()`	3 steps: `_fields_to_schema` → `_parse_transform` → `_generate_records`
Registry	`formats/`	Dynamic format registration, no cycles
Command	`cli.py`	Subparsers split by domain (core, files, network, web)

Metrics

Metric	Target	Status
Cyclomatic Complexity (avg)	≤ 2.5	In progress
Circular dependencies	0	✅ Achieved (formats/)
Pure functions in adapters	≥ 80%	✅ Achieved (FileSearchAdapter)
Test coverage	≥ 95%	✅ 96% (159 tests)

New Features (v0.3.0)

🎭 Faker Integration - Realistic Data

Generate real-world data (names, addresses, PESEL, NIP) using Faker:

pip install fraq[faker]

from fraq import generate

records = generate({
    'name': 'faker:pl_PL.name',      # Polish names
    'nip': 'faker:pl_PL.nip',         # Polish NIP
    'email': 'faker:email',
    'age': 'int:18-70',
}, count=1000, seed=42)

📊 DataFrame Export

Direct export to Polars, Pandas, or PyArrow:

pip install fraq[polars]  # or [pandas] or [arrow]

from fraq.dataframes import to_polars, to_pandas

# Polars DataFrame
df = to_polars({
    'sensor_id': 'str',
    'temperature': 'float:10-40',
    'humidity': 'float:0-100',
}, count=10000)

# Pandas DataFrame
df = to_pandas({...}, count=1000)

🧪 pytest Fixtures

Deterministic test data generation:

# conftest.py
from fraq.testing import fraq_fixture

@pytest.fixture
def users():
    return fraq_fixture({
        'user_id': 'str',
        'age': 'int:18-70',
        'active': 'bool',
    }, count=50, seed=42)

# test_users.py
def test_user_count(users):
    assert len(users) == 50
    assert all(18 <= u['age'] <= 70 for u in users)

🌀 IFS Generator - True Fractal Data

Competitors cannot replicate this! Generate data with structural self-similarity:

from fraq.ifs import create_ifs, OrganizationalMapper

# Create organizational hierarchy with fractal structure
ifs = create_ifs('organizational', seed=42)

data = ifs.generate(
    count=1000,
    depth=3,
    mapper=OrganizationalMapper(),
)

# Each level has statistically similar structure to parent

🔍 Fractal Schema Inference

Analyze real data → create fractal schema → generate infinite synthetic data:

from fraq.inference import infer_fractal
import pandas as pd

# Load real data
df = pd.read_csv('sales_data.csv')

# Infer fractal schema (no ML training needed!)
schema = infer_fractal(df.to_dict('records'))

# Generate synthetic data with same structure
synthetic = schema.generate(count=100000)

Advantages over SDV:

✅ No GPU required
✅ Deterministic (same input → same output)
✅ 1000x faster (no training)

📈 Benchmarks

Compare with competitors:

python -m fraq.benchmarks

Results:

Speed: fraq_stream 71,508 rec/s
Memory: fraq 139.7 MB (for 100k records)
Self-similarity: IFS > generate > Faker

text2fraq — Natural Language to Fractal Query

Convert natural language to fractal queries using LLM (LiteLLM) or rule-based fallback.

Setup (Ollama with small models)

# Install Ollama: https://ollama.com
ollama pull qwen2.5:3b      # Fast, instruction-tuned
ollama pull llama3.2:3b     # Balanced, multilingual
ollama pull phi3:3.8b       # Strong reasoning

# Install with AI extras
pip install -e ".[ai]"

# Copy and edit config
cp .env.example .env

Usage

from fraq import text2fraq, Text2Fraq, Text2FraqSimple

# One-liner (auto-fallback to rule-based if LLM unavailable)
result = text2fraq("Show 20 temperature readings in CSV")

# With specific model
from fraq.text2fraq import Text2FraqConfig
t2f = Text2Fraq(Text2FraqConfig(model="qwen2.5:3b"))
parsed = t2f.parse("Get deep analysis with depth 5")

# Rule-based (deterministic, no LLM needed)
parser = Text2FraqSimple()
query = parser.parse("Stream sensor data as JSON")

Environment Variables (.env)

Variable	Default	Description
`LITELLM_PROVIDER`	`ollama`	LLM provider
`LITELLM_MODEL`	`qwen2.5:3b`	Model name
`LITELLM_BASE_URL`	`http://localhost:11434`	API endpoint
`LITELLM_TEMPERATURE`	`0.1`	Generation temperature
`TEXT2FRAQ_DEFAULT_FORMAT`	`json`	Default output format
`TEXT2FRAQ_DEFAULT_DIMS`	`3`	Default fractal dimensions
`TEXT2FRAQ_DEFAULT_DEPTH`	`3`	Default query depth

Application Integrations

See examples/app_integrations.py for templates:

FastAPI — REST API with /query, /stream (SSE), /zoom/{depth}
Streamlit — Interactive dashboard with sliders and charts
Flask — Blueprints with NL endpoints
WebSocket — Real-time streaming server
Kafka — Producer/consumer with aiokafka
gRPC — High-performance RPC service
Celery — Background task processing
Jupyter — Interactive exploration widgets

Testing

pytest -v --cov=fraq    # 159 tests, 96% coverage

Project Structure

fraq/
├── fraq/
│   ├── __init__.py          # public API
│   ├── core.py              # FraqNode, FraqSchema, FraqCursor
│   ├── formats.py           # FormatRegistry + 6 built-in serialisers
│   ├── generators.py        # Hash, Fibonacci, Perlin, SensorStream
│   ├── query.py             # FraqQuery, FraqExecutor, FraqFilter
│   ├── adapters/            # File, HTTP, SQL, Sensor, Hybrid, Network, WebCrawler
│   │   ├── base.py
│   │   ├── file_adapter.py
│   │   ├── http_adapter.py
│   │   ├── sql_adapter.py
│   │   ├── sensor_adapter.py
│   │   ├── hybrid_adapter.py
│   │   ├── file_search.py
│   │   ├── network.py
│   │   ├── web_crawler.py
│   │   └── registry.py
│   ├── text2fraq/           # Natural language processing
│   │   ├── config.py
│   │   ├── models.py
│   │   ├── llm_client.py
│   │   ├── parser_rules.py
│   │   ├── parser_llm.py
│   │   ├── router.py        # v0.2.10: ModelRouter
│   │   ├── session.py       # v0.2.10: FraqSession
│   │   └── shortcuts.py
│   ├── schema_export.py     # NLP2CMD, OpenAPI, GraphQL, AsyncAPI, Proto, JSON Schema
│   ├── streaming.py         # AsyncFraqStream, async_query, async_stream
│   ├── server.py            # v0.2.10: FastAPI production server
│   └── cli.py               # CLI entry point
├── tests/                   # 159 test suite
├── examples/
│   ├── basic/               # Core examples
│   ├── text2fraq/           # NLP examples
│   ├── network/             # Network & web crawling
│   ├── v028/                # v0.2.10 new features
│   └── *-docker/            # Docker compositions
└── .github/workflows/       # CI/CD

License

Apache License 2.0 - see LICENSE for details.

Author

Created by Tom Sapletta - tom@sapletta.com

Name		Name	Last commit message	Last commit date
Latest commit History 25 Commits
.github/workflows		.github/workflows
docs		docs
examples		examples
fraq		fraq
project		project
tests		tests
.code2docs.api_snapshot.json		.code2docs.api_snapshot.json
.env.example		.env.example
.gitignore		.gitignore
.pre-commit-config.yaml		.pre-commit-config.yaml
CHANGELOG.md		CHANGELOG.md
Dockerfile		Dockerfile
Dockerfile.cli		Dockerfile.cli
Dockerfile.websocket		Dockerfile.websocket
LICENSE		LICENSE
MANIFEST.in		MANIFEST.in
PKG-INFO		PKG-INFO
PLAN_REFACTORING.md		PLAN_REFACTORING.md
README.md		README.md
VERSION		VERSION
code2docs.yaml		code2docs.yaml
docker-compose.yml		docker-compose.yml
goal.yaml		goal.yaml
main_websocket.py		main_websocket.py
project.sh		project.sh
pyproject.toml		pyproject.toml
setup.cfg		setup.cfg

Folders and files

Latest commit

History

Repository files navigation

fraq — Fractal Query Data Library

Core Idea

Key Properties

Quick Start

Python API

🆕 NEW: Simplified API (v0.2.11+)

🆕 DataFrame Export (v0.2.11+)

🆕 IFS Fractal Generator (v0.2.11+)

🆕 Faker Integration (v0.2.11+)

🆕 pytest Integration (v0.2.11+)

Classic API

NLP2CMD Integration

New in v0.2.10 — Multi-Model Router & Session Memory

NLP2CMD Integration

Schema Export Formats

Generators

Data Source Adapters

Architecture

Code Organization

Design Patterns

Metrics

New Features (v0.3.0)

🎭 Faker Integration - Realistic Data

📊 DataFrame Export

🧪 pytest Fixtures

🌀 IFS Generator - True Fractal Data

🔍 Fractal Schema Inference

📈 Benchmarks

text2fraq — Natural Language to Fractal Query

Setup (Ollama with small models)

Usage

Environment Variables (.env)

Application Integrations

Testing

Project Structure

License

Author

About

Resources

License

Contributing

Uh oh!

Stars

Watchers

Forks

Releases

Packages 0

Uh oh!

Contributors

Uh oh!

Languages

Packages