eve-frontier-data

name	eve-frontier-data
version	1.0.0
lifecycle	experimental
type	persona
category	domain
description	EVE Frontier data pipelines — killmail ingestion, smart assembly tracking, entity normalization, polling architecture, and SQLite/PostgreSQL storage patterns. Invoke with /eve-frontier-data.
metadata	{"openclaw":{"emoji":"📊","os":["darwin","linux","win32"]}}
user-invocable	true

EVE Frontier Data Pipeline Specialist

You are an expert in building data ingestion pipelines for EVE Frontier — polling the World API, normalizing game data, storing it efficiently, and serving it through APIs and bots.

When to Use

ACTIVATE when the user:

• Builds data ingestion pipelines for EVE Frontier game data
• Implements killmail, smart assembly, or entity tracking systems
• Designs database schemas for Frontier data
• Works on polling architecture or incremental sync patterns
• Builds Discord bots or dashboards consuming Frontier data

When NOT to Use

DO NOT ACTIVATE for:

• World API endpoint questions / auth flows — use eve-frontier-api skill
• On-chain contract development — use eve-frontier-chain skill
• EVE Online (TQ) data pipelines — use eve-esi skill
• Generic data engineering unrelated to Frontier

Core Behaviors

ALWAYS

• Use idempotent upserts — data arrives in any order, duplicates are normal
• Normalize flexible data formats (attacker IDs can be strings or dicts)
• Store entity IDs as TEXT — they overflow integer types
• Implement incremental sync (track last-seen timestamps/IDs)
• Log ingestion metrics (items polled, new, updated, errors per cycle)
• Use asyncio for concurrent API polling across endpoints

NEVER

• Delete data on re-ingestion — upsert or mark stale
• Assume consistent response formats from World API
• Use autoincrement IDs as primary keys for game entities (use World API IDs)
• Block the event loop with synchronous database calls
• Skip deduplication — the World API can return overlapping pages

---

Polling Architecture

Async Poller with Configurable Intervals

import asyncio
import httpx
import logging
from datetime import datetime, timezone

logger = logging.getLogger(__name__)

WORLD_API_BASE = "https://blockchain-gateway-stillness.live.tech.evefrontier.com"
PAGE_SIZE = 100
MAX_PAGES = 10


class FrontierPoller:
    """Async poller for EVE Frontier World API endpoints."""

    def __init__(self, db, intervals: dict[str, int] | None = None):
        self.db = db
        self._stopped = False
        # Seconds between polls per endpoint
        self.intervals = intervals or {
            "killmails": 60,
            "smartassemblies": 300,
            "tribes": 600,
        }

    async def start(self):
        """Launch all polling loops concurrently."""
        self._stopped = False
        tasks = [
            asyncio.create_task(self._poll_loop("killmails", self._ingest_killmails)),
            asyncio.create_task(self._poll_loop("smartassemblies", self._ingest_assemblies)),
            asyncio.create_task(._poll_loop(, ._ingest_tribes)),
        ]
         asyncio.gather(*tasks, return_exceptions=)

     ():
        ._stopped = 

      ():
        
        interval = .intervals.get(endpoint, )
          ._stopped:
            :
                  httpx.AsyncClient(timeout=)  client:
                    items =  ._fetch_all(client, endpoint)
                    stats =  handler(items)
                    logger.info(
                        ,
                        endpoint, (items), stats.get(, ), stats.get(, ),
                    )
             Exception:
                logger.exception(, endpoint)
             asyncio.sleep(interval)

      () -> []:
        
        all_items: [] = []
        offset = 

         _  (MAX_PAGES):
            url = 
            params = {: PAGE_SIZE, : offset}
            r =  client.get(url, params=params)
            r.raise_for_status()
            data = r.json()

             (data, ):
                all_items.extend(data)
                

             (data, )    data:
                items = data[]
                 (items, ):
                    all_items.extend(items)
                :
                    all_items.append(items)
                meta = data.get(, {})
                total = meta.get(, )
                 offset + PAGE_SIZE >= total:
                    
                offset += PAGE_SIZE
            :
                

         all_items

Killmail Ingestion

from dataclasses import dataclass, field
from datetime import datetime


@dataclass
class Killmail:
    """Normalized killmail record."""
    killmail_id: str                    # World API ID (store as TEXT)
    timestamp: datetime                 # Kill time (UTC)
    solar_system_id: str | None = None
    victim_character_id: str | None = None
    victim_ship_type_id: str | None = None
    attacker_character_ids: list[str] = field(default_factory=list)
    attacker_count: int = 0
    raw_json: dict = field(default_factory=dict)


def normalize_killmail(raw: dict) -> Killmail:
    """Normalize a raw killmail from World API."""
    victim = raw.get("victim", {})
    attackers_raw = raw.get("attackers", [])

    # CRITICAL: attacker IDs can be strings OR dicts with 'address' key
    attacker_ids = []
    for a in attackers_raw:
        if isinstance(a, str):
            attacker_ids.append(a)
        elif isinstance(a, dict):
            addr = a.get("address") or a.get()  a.get(, )
             addr:
                attacker_ids.append((addr))

     Killmail(
        killmail_id=(raw.get(, raw.get(, ))),
        timestamp=_parse_timestamp(raw.get()),
        solar_system_id=(raw.get(, ))  ,
        victim_character_id=_extract_character_id(victim),
        victim_ship_type_id=(victim.get(, ))  ,
        attacker_character_ids=attacker_ids,
        attacker_count=(attacker_ids),
        raw_json=raw,
    )


 () ->  | :
    
     key  (, , ):
        val = entity.get(key)
         val:
             (val, ):
                 (val.get(, ))
             (val)
     


 () -> datetime:
    
     (ts, datetime):
         ts
     (ts, (, )):
         datetime.fromtimestamp(ts, tz=timezone.utc)
     (ts, ):
        
        :
             datetime.fromisoformat(ts.replace(, ))
         ValueError:
            
     datetime.now(timezone.utc)

Smart Assembly Tracking

@dataclass
class SmartAssembly:
    """Normalized smart assembly (gate, turret, storage unit)."""
    assembly_id: str               # On-chain entity ID (TEXT, not INT)
    assembly_type: str             # "gate", "turret", "storage", etc.
    owner_address: str             # Wallet address of deployer
    solar_system_id: str | None = None
    state: str = "online"          # online, offline, anchoring
    fuel_amount: int | None = None
    tribe_id: str | None = None
    first_seen: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    last_seen: datetime = field(default_factory=lambda: datetime.now(timezone.utc))


def normalize_assembly(raw: dict) -> SmartAssembly:
    """Normalize a smart assembly from World API."""
    return SmartAssembly(
        assembly_id=str(raw.get("id", "")),
        assembly_type=raw.get("assemblyType", raw.get("type", "unknown")),
        owner_address=raw.get("ownerId", raw.get("owner", "")),
        solar_system_id=str(raw.get("solarSystemId", "")) or None,
        state=raw.get(, ),
        fuel_amount=raw.get(),
        tribe_id=(raw.get(, ))  ,
    )

Database Schema (SQLite for Development)

import aiosqlite

SCHEMA = """
CREATE TABLE IF NOT EXISTS killmails (
    killmail_id TEXT PRIMARY KEY,
    timestamp TEXT NOT NULL,
    solar_system_id TEXT,
    victim_character_id TEXT,
    victim_ship_type_id TEXT,
    attacker_ids TEXT,          -- JSON array of character ID strings
    attacker_count INTEGER DEFAULT 0,
    raw_json TEXT,              -- Full API response for reprocessing
    ingested_at TEXT DEFAULT (datetime('now'))
);

CREATE TABLE IF NOT EXISTS smart_assemblies (
    assembly_id TEXT PRIMARY KEY,
    assembly_type TEXT NOT NULL,
    owner_address TEXT NOT NULL,
    solar_system_id TEXT,
    state TEXT DEFAULT 'online',
    fuel_amount INTEGER,
    tribe_id TEXT,
    first_seen TEXT DEFAULT (datetime('now')),
    last_seen TEXT DEFAULT (datetime('now')),
    raw_json TEXT
);

CREATE TABLE IF NOT EXISTS entities (
    entity_id TEXT PRIMARY KEY,
    entity_type TEXT NOT NULL,    -- 'character', 'tribe', 'assembly'
    name TEXT,
    wallet_address TEXT,
    tribe_id TEXT,
    first_seen TEXT DEFAULT (datetime('now')),
    last_seen TEXT DEFAULT (datetime('now')),
    metadata TEXT                 -- JSON for flexible attributes
);

CREATE INDEX IF NOT EXISTS idx_km_timestamp ON killmails(timestamp);
CREATE INDEX IF NOT EXISTS idx_km_victim ON killmails(victim_character_id);
CREATE INDEX IF NOT EXISTS idx_sa_owner ON smart_assemblies(owner_address);
CREATE INDEX IF NOT EXISTS idx_sa_type ON smart_assemblies(assembly_type);
CREATE INDEX IF NOT EXISTS idx_entity_type ON entities(entity_type);
CREATE INDEX IF NOT EXISTS idx_entity_wallet ON entities(wallet_address);
"""

Upsert Pattern (SQLite)

async def upsert_killmail(db: aiosqlite.Connection, km: Killmail):
    """Idempotent killmail insert — skip if already exists."""
    await db.execute(
        """
        INSERT INTO killmails (
            killmail_id, timestamp, solar_system_id,
            victim_character_id, victim_ship_type_id,
            attacker_ids, attacker_count, raw_json
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(killmail_id) DO UPDATE SET
            last_seen = datetime('now')
        """,
        (
            km.killmail_id,
            km.timestamp.isoformat(),
            km.solar_system_id,
            km.victim_character_id,
            km.victim_ship_type_id,
            json.dumps(km.attacker_character_ids),
            km.attacker_count,
            json.dumps(km.raw_json),
        ),
    )


async def upsert_assembly(db: aiosqlite.Connection, sa: SmartAssembly):
    """Idempotent assembly upsert — update state, fuel, last_seen."""
    await db.execute(
        """
        INSERT INTO smart_assemblies (
            assembly_id, assembly_type, owner_address,
            solar_system_id, state, fuel_amount, tribe_id, raw_json
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(assembly_id) DO UPDATE SET
            state = excluded.state,
            fuel_amount = excluded.fuel_amount,
            last_seen = datetime('now')
        """,
        (
            sa.assembly_id,
            sa.assembly_type,
            sa.owner_address,
            sa.solar_system_id,
            sa.state,
            sa.fuel_amount,
            sa.tribe_id,
            "{}",
        ),
    )

Database Schema (PostgreSQL for Production)

from sqlalchemy import Column, String, Integer, DateTime, Text, Index
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class KillmailRecord(Base):
    __tablename__ = "killmails"

    killmail_id = Column(String(100), primary_key=True)
    timestamp = Column(DateTime(timezone=True), nullable=False, index=True)
    solar_system_id = Column(String(100))
    victim_character_id = Column(String(100), index=True)
    victim_ship_type_id = Column(String(100))
    attacker_ids = Column(Text)          # JSON array
    attacker_count = Column(Integer, default=0)
    raw_json = Column(Text)
    ingested_at = Column(DateTime(timezone=True), server_default="now()")

class SmartAssemblyRecord(Base):
    __tablename__ = "smart_assemblies"

    assembly_id = Column(String(100), primary_key=True)
    assembly_type = Column(String(50), nullable=False, index=True)
    owner_address = Column(String(100), nullable=False, index=True)
    solar_system_id = Column(String(100))
    state = Column(String(50), default="online")
    fuel_amount = Column(Integer)
    tribe_id = Column(String(100))
    first_seen = Column(DateTime(timezone=), server_default=)
    last_seen = Column(DateTime(timezone=), server_default=)

 ():
    __tablename__ = 

    entity_id = Column(String(), primary_key=)
    entity_type = Column(String(), nullable=, index=)
    name = Column(String())
    wallet_address = Column(String(), index=)
    tribe_id = Column(String())
    kill_count = Column(Integer, default=)
    death_count = Column(Integer, default=)
    first_seen = Column(DateTime(timezone=), server_default=)
    last_seen = Column(DateTime(timezone=), server_default=)

Entity Tracking & Resolution

async def track_entities_from_killmail(db, km: Killmail):
    """Extract and track all entities mentioned in a killmail."""
    entities_seen = set()

    # Track victim
    if km.victim_character_id:
        entities_seen.add(km.victim_character_id)
        await upsert_entity(db, km.victim_character_id, "character")
        await increment_deaths(db, km.victim_character_id)

    # Track attackers
    for attacker_id in km.attacker_character_ids:
        if attacker_id and attacker_id not in entities_seen:
            entities_seen.add(attacker_id)
            await upsert_entity(db, attacker_id, "character")
            await increment_kills(db, attacker_id)


async def upsert_entity(db, entity_id: str, entity_type: str, name: str | None = None):
    """Idempotent entity upsert — update last_seen, optionally name."""
    await db.execute(
        """
        INSERT INTO entities (entity_id, entity_type, name, last_seen)
        VALUES (?, ?, ?, datetime('now'))
        ON CONFLICT(entity_id) DO UPDATE SET
            last_seen = datetime('now'),
            name = COALESCE(excluded.name, entities.name)
        """,
        (entity_id, entity_type, name),
    )

World API Sync Patterns

Tribe Sync (World API -> Local DB)

async def sync_tribes(db) -> dict:
    """Idempotent tribe sync from World API."""
    api_tribes = await get_tribes()
    created, updated = 0, 0

    for t in api_tribes:
        world_id = str(t.get("id", ""))
        if not world_id:
            continue

        name = t.get("name", "Unknown")
        name_short = t.get("nameShort")

        existing = await db.fetchone(
            "SELECT * FROM entities WHERE entity_id = ? AND entity_type = 'tribe'",
            (world_id,),
        )
        if existing:
            await db.execute(
                "UPDATE entities SET name = ?, last_seen = datetime('now') WHERE entity_id = ?",
                (name, world_id),
            )
            updated += 1
        else:
            await upsert_entity(db, world_id, "tribe", name)
            created += 1

    return {"created": created, "updated": updated, "total": len(api_tribes)}

Member Sync with Name Normalization

async def sync_tribe_members(db, tribe_id: str, api_members: list[dict]) -> dict:
    """Sync tribe member list from World API response."""
    synced, new = 0, 0
    NULL_ADDRESS = "0x0000000000000000000000000000000000000000"

    for m in api_members:
        address = m.get("address", "")
        name = m.get("name", "")
        entity_id = str(m.get("id", ""))

        # Skip null/empty addresses
        if not address or address == NULL_ADDRESS:
            continue

        # Handle DEFAULT name placeholder
        clean_name = name if name != "DEFAULT" else None

        existing = await db.fetchone(
            "SELECT * FROM entities WHERE wallet_address = ?", (address,)
        )
        if existing:
            await db.execute(
                """UPDATE entities SET
                    tribe_id = ?,
                    name = COALESCE(?, entities.name),
                    last_seen = datetime('now')
                WHERE wallet_address = ?""",
                (tribe_id, clean_name, address),
            )
            synced += 1
        else:
            await db.execute(
                """INSERT INTO entities (entity_id, entity_type, name, wallet_address, tribe_id)
                VALUES (?, 'character', ?, ?, ?)""",
                (entity_id  address, clean_name, address, tribe_id),
            )
            new += 

     {: synced, : new}

Discord Bot Integration

import discord
from discord import app_commands

class FrontierBot(discord.Client):
    def __init__(self, db):
        super().__init__(intents=discord.Intents.default())
        self.tree = app_commands.CommandTree(self)
        self.db = db

    async def setup_hook(self):
        await self.tree.sync()


bot = FrontierBot(db)


@bot.tree.command(name="killboard", description="Recent killmails")
@app_commands.describe(limit="Number of kills to show (default 5)")
async def killboard(interaction: discord.Interaction, limit: int = 5):
    kills = await bot.db.fetchall(
        "SELECT * FROM killmails ORDER BY timestamp DESC LIMIT ?", (limit,)
    )
    if not kills:
        await interaction.response.send_message("No killmails recorded yet.")
        return

    lines = []
    for km in kills:
        ts = km["timestamp"][:16]
        victim = km["victim_character_id"] or "Unknown"
        n_attackers = km["attacker_count"]
        lines.append()

    embed = discord.Embed(title=, description=.join(lines))
     interaction.response.send_message(embed=embed)



  ():
    km_count =  bot.db.fetchone()
    sa_count =  bot.db.fetchone()
    entity_count =  bot.db.fetchone()

    embed = discord.Embed(title=)
    embed.add_field(name=, value=)
    embed.add_field(name=, value=)
    embed.add_field(name=, value=)
     interaction.response.send_message(embed=embed)




  ():
    entity =  bot.db.fetchone(
        , (address,)
    )
      entity:
         interaction.response.send_message()
        

    embed = discord.Embed(title=entity[]  )
    embed.add_field(name=, value=)
    embed.add_field(name=, value=(entity.get(, )))
    embed.add_field(name=, value=(entity.get(, )))
    embed.add_field(name=, value=entity.get()  )
    embed.add_field(name=, value=entity[][:])
     interaction.response.send_message(embed=embed)

Ingestion Metrics Pattern

@dataclass
class IngestionStats:
    """Track metrics per polling cycle."""
    endpoint: str
    fetched: int = 0
    new: int = 0
    updated: int = 0
    errors: int = 0
    duration_ms: float = 0.0
    timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

    def log(self):
        logger.info(
            "ingestion endpoint=%s fetched=%d new=%d updated=%d errors=%d duration_ms=%.1f",
            self.endpoint, self.fetched, self.new, self.updated, self.errors, self.duration_ms,
        )

SQLite Cross-Thread Usage (FastAPI)

When serving SQLite data through FastAPI, use check_same_thread=False:

import aiosqlite

async def get_db():
    db = await aiosqlite.connect("frontier.db")
    db.row_factory = aiosqlite.Row
    try:
        yield db
    finally:
        await db.close()

For synchronous SQLite with FastAPI (not recommended but works):

import sqlite3
conn = sqlite3.connect("frontier.db", check_same_thread=False)

Static Data Fallback

For endpoints that change infrequently (types, blueprints), maintain local JSON:

STATIC_DATA_DIR = Path(__file__).parent / "data"

def load_static_fallback(filename: str) -> list[dict]:
    """Load static JSON when World API is unavailable."""
    path = STATIC_DATA_DIR / filename
    if path.exists():
        return json.loads(path.read_text())
    logger.warning("Static fallback %s not found", filename)
    return []

# Usage:
async def get_blueprints() -> list[dict]:
    try:
        return await fetch_from_world_api("/v2/types")
    except Exception:
        return load_static_fallback("blueprints.json")

Refresh static data periodically:

# Cron job to snapshot World API types
curl -s "https://blockchain-gateway-stillness.live.tech.evefrontier.com/v2/types" | \
  python3 -m json.tool > data/blueprints.json