chore: remove tracked __pycache__ artifacts, add gitignore

- evm.py: map MintSwapV2/RedeemSwapV2/CreateNewMarket; decode indexed address topics correctly
- store.py: parse MintSwapV2/RedeemSwapV2 fields, compute real trade price (no [0,1] clamp), dirty-queue for trades/klines
- repository.py: asyncpg upsert of 5s OHLCV bars + trades
- main.py: watch controller/vault-factory/default market, dynamic market discovery, checkpoint resume
- checkpoint.py: JSON checkpoint persistence

.gitignore

@@ -0,0 +1,6 @@
+__pycache__/
+*.pyc
+.venv/
+.env
+data/
+checkpoint_*.json

ARCHITECTURE.md

@@ -0,0 +1,73 @@
+# WTFX Indexer Architecture & Principles (索引器架构与核心铁律)
+
+> **定位**：Indexer 不是数据库搬运工，而是 **“区块链事实 → WTFX 可查询状态” 的确定性投影系统**。
+> **运行形态**：第一阶段单进程运行，但严格按照 **Fetch -> Decode -> Process -> Project** 四层边界设计，支持**零停机热追高、全量可重放、幂等去重、防链重组 (Reorg)**。
+
+---
+
+## 1. 五大核心架构铁律 (Non-Negotiable Principles)
+
+1. **【三层数据流转：Raw -> Normalized -> Projection】**：
+   - **Raw Layer**：原样保存链上原始日志 (`raw_logs`, `raw_blocks`)。
+   - **Normalized Layer**：解析为跨链统一的标准事件字典 (`chain_events`: `ORDER_FILLED`, `MARKET_CREATED`, `SETTLEMENT`)。
+   - **Projection Layer**：生成业务最终可查询的只读视图 (`markets`, `trades`, `positions`, `token_balances`)。
+2. **【绝对可重放性 (Replayability)】**：
+   - 当业务投影算法调整或修复时，**严禁手工修数据库**；
+   - 必须支持：清空 Projection 视图 $␍ightarrow$ 从指定 Checkpoint 或 Block 0 重新重放 Normalized Events $␍ightarrow$ 确定性重新生成 Projection。
+3. **【幂等去重保证 (Idempotency)】**：
+   - 每个事件唯一凭证为 `(chain_id, tx_hash, log_index)`；
+   - 数据库唯一约束拦截重复写入，重复消费或追高扫描时无脑跳过已存在事件。
+4. **【安全检查点与防重组 (Checkpoint & Reorg Protection)】**：
+   - 维护持久化 `indexer_state` 记录安全落盘水位；
+   - 区分 `CONFIRMED`（快速响应）与 `FINALIZED`（安全落盘），为最新区块预留回退重测窗口。
+5. **【不直接越权控制业务资金】**：
+   - Indexer 只负责**“发现链上发生了什么”**，严禁直接在 Indexer 内部修改用户余额；
+   - 资金结算与账本生成交由 Backend 的 Settlement Service 处理。
+
+---
+
+## 2. 目录结构全景
+
+```
+wtf-indexer/
+├── src/
+│   ├── main.py                     # 索引器服务入口与管道调度器 (Pipeline Runner)
+│   ├── config.py                   # RPC、链配置、数据库与扫描区间配置
+│   │
+│   ├── core/                       # 核心抽象与不可变数据模型
+│   │   ├── events.py               # 规范化事件定义 (Normalized Event Schema)
+│   │   ├── checkpoint.py           # Checkpoint 水位管理器与 Reorg 检测
+│   │   └── models.py               # 基础类型与接口抽象
+│   │
+│   ├── ingestion/                  # 链上数据拉取层 (Fetcher)
+│   │   ├── fetcher.py              # RPC 批量抓块与重试调度 (Batch eth_getLogs)
+│   │   └── queue.py                # 内存/Redis 缓冲队列
+│   │
+│   ├── decoding/                   # 事件解析层 (Decoder)
+│   │   ├── evm.py                  # EVM ABI 日志解码器 (WTFMarketV2, Controller)
+│   │   └── base.py                 # 解码器抽象基类
+│   │
+│   ├── processors/                 # 业务领域处理器 (Processor)
+│   │   ├── market.py               # 市场创建/状态流转处理
+│   │   ├── trading.py              # 订单成交与代币铸造/赎回处理
+│   │   └── settlement.py           # 到期裁决与兑付结算处理
+│   │
+│   ├── projections/                # 确定性投影层 (Projection Views)
+│   │   ├── markets.py              # 市场聚合状态表投影
+│   │   ├── trades.py               # 逐笔交易记录投影
+│   │   └── positions.py            # 用户持仓与成本投影
+│   │
+│   ├── chains/                     # 链适配器 (Chain Adapters)
+│   │   ├── robinhood.py            # Robinhood Chain (Testnet/Mainnet) 适配器
+│   │   └── base.py                 # 通用 EVM 适配器
+│   │
+│   └── infrastructure/             # 底层设施
+│       ├── postgres.py             # 异步 PostgreSQL 连接与投影持久化
+│       ├── redis.py                # Redis 锁与事件总线广播
+│       └── rpc.py                  # 健壮的多节点 RPC 客户端
+│
+├── tests/                          # 可重放测试与解码单元测试
+├── Dockerfile                      # 容器构建镜像
+├── requirements.txt                # 依赖包配置
+└── ARCHITECTURE.md                 # 架构铁律
+```

Dockerfile

@@ -0,0 +1,16 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    libpq-dev \
+    curl \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+CMD ["python", "-m", "src.main"]

requirements.txt

@@ -0,0 +1,9 @@
+web3>=6.19.0
+eth-abi>=5.1.0
+pydantic>=2.7.0
+pydantic-settings>=2.3.0
+asyncpg>=0.29.0
+sqlalchemy[asyncio]>=2.0.30
+redis>=5.0.4
+pytest>=8.2.0
+pytest-asyncio>=0.23.7

src/config.py

@@ -0,0 +1,29 @@
+from pydantic_settings import BaseSettings
+
+class IndexerSettings(BaseSettings):
+    CHAIN_ID: int = 46630 # Robinhood Testnet
+    RPC_URL: str = "https://rpc.testnet.chain.robinhood.com"
+    # 回放起点：默认为测试网控制器/market 部署后的区块，可通过环境变量覆盖
+    START_BLOCK: int = 110000000
+    BATCH_SIZE: int = 500
+    POLL_INTERVAL_SECONDS: float = 2.0
+    CONFIRMATIONS: int = 1 # Testnet 确认数
+
+    # 数据库与 Redis
+    DATABASE_URL: str = "postgresql+asyncpg://wtfx_user:wtfx_pass@localhost:5432/wtfx_db"
+    REDIS_URL: str = "redis://localhost:6379/0"
+
+    # checkpoint 持久化文件
+    CHECKPOINT_FILE: str = "./data/checkpoint_46630.json"
+
+    # 已部署合约地址 (Robinhood Testnet)
+    CONTROLLER_ADDRESS: str = "0xc0E24E152771C588B21AEB654b30B1cBAf381c1a"
+    COLLATERAL_ADDRESS: str = "0xe776e957953EA69b7Eaa9d7d4098aBC076bDD5E7"
+    CURVE_ADDRESS: str = "0xF0E189974c413506098AFB6Bc6Bf4C6715fBf7B1"
+    VAULT_FACTORY_ADDRESS: str = "0x89401e07296267c01017cA150D5Ec8883a78e0B2"
+
+    class Config:
+        env_file = ".env"
+        extra = "ignore"
+
+indexer_settings = IndexerSettings()

src/core/abis.py

@@ -0,0 +1,22 @@
+import json
+import os
+from typing import Dict, Any
+
+def load_contract_abi(name: str) -> list:
+    """加载共享合约 ABI"""
+    # 优先从 frontend packages 读取
+    possible_paths = [
+        os.path.join(os.path.dirname(__file__), "..", "..", "..", "wtf-frontend", "packages", "contracts", "src", "abis", f"{name}.json"),
+        os.path.join(os.path.dirname(__file__), "..", "..", "..", "wtf-contract", "artifacts", "main", "src", "controllerv2", f"{name}.sol", f"{name}.json"),
+        os.path.join(os.path.dirname(__file__), "..", "..", "..", "wtf-contract", "artifacts", "main", "src", "marketv2", f"{name}.sol", f"{name}.json")
+    ]
+    for p in possible_paths:
+        if os.path.exists(p):
+            with open(p, "r", encoding="utf-8") as f:
+                data = json.load(f)
+                return data.get("abi", data) if isinstance(data, dict) else data
+    return []
+
+CONTROLLER_ABI = load_contract_abi("WTFControllerV2")
+MARKET_ABI = load_contract_abi("WTFMarketV2")
+MOCK_ERC20_ABI = load_contract_abi("MockERC20")

src/core/checkpoint.py

@@ -0,0 +1,38 @@
+import json
+import logging
+import os
+from pathlib import Path
+from typing import Optional
+
+from pydantic import BaseModel
+
+logger = logging.getLogger(__name__)
+
+class CheckpointState(BaseModel):
+    chain_id: int
+    last_processed_block: int
+    last_processed_tx: str = ""
+    updated_at: int = 0
+
+    @classmethod
+    def load(cls, file_path: str, chain_id: int, default_block: int) -> "CheckpointState":
+        try:
+            p = Path(file_path)
+            if p.exists():
+                data = json.loads(p.read_text(encoding="utf-8"))
+                state = cls(**data)
+                if state.chain_id == chain_id:
+                    return state
+        except Exception as e:
+            logger.warning(f"Failed to load checkpoint from {file_path}: {e}")
+        return cls(chain_id=chain_id, last_processed_block=default_block)
+
+    def save(self, file_path: str):
+        try:
+            p = Path(file_path)
+            p.parent.mkdir(parents=True, exist_ok=True)
+            tmp = p.with_suffix(".tmp")
+            tmp.write_text(self.model_dump_json(), encoding="utf-8")
+            os.replace(tmp, p)
+        except Exception as e:
+            logger.warning(f"Failed to save checkpoint to {file_path}: {e}")

src/core/events.py

@@ -0,0 +1,18 @@
+from pydantic import BaseModel, Field
+from typing import Dict, Any, Optional
+
+class NormalizedEvent(BaseModel):
+    """跨链与跨合约的标准归一化事件 (Layer 2)"""
+    chain_id: int
+    block_number: int
+    tx_hash: str
+    log_index: int
+    event_type: str = Field(description="e.g. MARKET_CREATED, MINT_COLLATERAL, REDEEM_OT, RESOLVE_OUTCOME, CLAIM")
+    contract_address: str
+    payload: Dict[str, Any]
+    timestamp: int
+
+    @property
+    def event_id(self) -> str:
+        """全局唯一幂等标识 (Idempotency Key)"""
+        return f"{self.chain_id}:{self.tx_hash}:{self.log_index}"

src/decoding/evm.py

@@ -0,0 +1,153 @@
+import json
+import logging
+from typing import Dict, Any, Optional
+from web3 import Web3
+from eth_abi import decode
+from src.core.events import NormalizedEvent
+from src.core.abis import CONTROLLER_ABI, MARKET_ABI
+
+logger = logging.getLogger(__name__)
+
+# Vault Factory & User Vault Events ABI
+VAULT_EVENTS_ABI = [
+    {
+        "anonymous": False,
+        "inputs": [
+            {"indexed": True, "name": "user", "type": "address"},
+            {"indexed": True, "name": "vault", "type": "address"},
+            {"indexed": False, "name": "vaultIndex", "type": "uint256"}
+        ],
+        "name": "VaultCreated",
+        "type": "event"
+    },
+    {
+        "anonymous": False,
+        "inputs": [
+            {"indexed": True, "name": "token", "type": "address"},
+            {"indexed": True, "name": "from", "type": "address"},
+            {"indexed": False, "name": "amount", "type": "uint256"}
+        ],
+        "name": "Deposited",
+        "type": "event"
+    },
+    {
+        "anonymous": False,
+        "inputs": [
+            {"indexed": True, "name": "token", "type": "address"},
+            {"indexed": True, "name": "to", "type": "address"},
+            {"indexed": False, "name": "amount", "type": "uint256"}
+        ],
+        "name": "Withdrawn",
+        "type": "event"
+    }
+]
+
+class EVMDecoder:
+    """EVM ABI 日志规范化解码器 (包含 Controller, Market 与 User Vault)"""
+
+    def __init__(self, chain_id: int):
+        self.chain_id = chain_id
+        self.w3 = Web3()
+        self._build_topic_maps()
+
+    def _build_topic_maps(self):
+        self.event_abi_map = {}
+        
+        for abi in CONTROLLER_ABI + MARKET_ABI + VAULT_EVENTS_ABI:
+            if abi.get("type") == "event":
+                name = abi.get("name")
+                inputs = abi.get("inputs", [])
+                types = [i["type"] for i in inputs]
+                sig = f"{name}({','.join(types)})"
+                topic0 = self.w3.keccak(text=sig).hex()
+                self.event_abi_map[topic0] = abi
+
+    def decode_log(self, log: Dict[str, Any], block_timestamp: int = 0) -> Optional[NormalizedEvent]:
+        topics = log.get("topics", [])
+        if not topics:
+            return None
+
+        topic0 = topics[0].hex() if isinstance(topics[0], bytes) else topics[0]
+        abi = self.event_abi_map.get(topic0)
+        if not abi:
+            return None
+
+        event_name = abi["name"]
+        contract_addr = log.get("address", "").lower()
+        tx_hash = log.get("transactionHash", "").hex() if isinstance(log.get("transactionHash"), bytes) else str(log.get("transactionHash", ""))
+        block_number = log.get("blockNumber", 0)
+        log_index = log.get("logIndex", 0)
+
+        payload: Dict[str, Any] = {}
+        indexed_inputs = [i for i in abi.get("inputs", []) if i.get("indexed")]
+        non_indexed_inputs = [i for i in abi.get("inputs", []) if not i.get("indexed")]
+
+        for idx, inp in enumerate(indexed_inputs):
+            if idx + 1 < len(topics):
+                raw_topic = topics[idx + 1]
+                if isinstance(raw_topic, bytes):
+                    t_bytes = raw_topic
+                else:
+                    t_hex = raw_topic[2:] if raw_topic.startswith("0x") else raw_topic
+                    t_bytes = bytes.fromhex(t_hex)
+                if inp["type"] == "address":
+                    # indexed address topic 右对齐 20 字节，需要去掉前导 0 并转成标准 0x 地址
+                    payload[inp["name"]] = self.w3.to_checksum_address("0x" + t_bytes[-20:].hex())
+                else:
+                    payload[inp["name"]] = "0x" + t_bytes.hex()
+
+        data = log.get("data", "0x")
+        if isinstance(data, str) and data.startswith("0x"):
+            data_bytes = bytes.fromhex(data[2:])
+        elif isinstance(data, bytes):
+            data_bytes = data
+        else:
+            data_bytes = b""
+
+        if data_bytes and non_indexed_inputs:
+            types = [i["type"] for i in non_indexed_inputs]
+            try:
+                decoded_vals = decode(types, data_bytes)
+                for inp, val in zip(non_indexed_inputs, decoded_vals):
+                    if isinstance(val, bytes):
+                        payload[inp["name"]] = "0x" + val.hex()
+                    else:
+                        payload[inp["name"]] = val
+            except Exception as e:
+                logger.error(f"Failed to decode non-indexed data for {event_name}: {e}")
+
+        type_mapping = {
+            # WTFControllerV2 / MarketFactory
+            "CreateNewMarket": "MARKET_DEPLOYED",
+            "CreateNewQuestionV2": "QUESTION_CREATED",
+            "Resolve": "OUTCOME_RESOLVED",
+            "Unresolve": "OUTCOME_UNRESOLVED",
+            "Finalise": "MARKET_FINALISED",
+            "OverrideFinalise": "OUTCOME_OVERRIDDEN",
+            "ManuallyFinalise": "MARKET_FINALISED",
+            # WTFMarketV2 (V2 真实成交事件)
+            "MintSwapV2": "ORDER_MINT",
+            "RedeemSwapV2": "ORDER_REDEEM",
+            "MintSwap": "ORDER_MINT",
+            "RedeemSwap": "ORDER_REDEEM",
+            "ClaimPayout": "POSITION_CLAIMED",
+            "GraduateMarket": "MARKET_GRADUATED",
+            "MarketRefunded": "MARKET_REFUNDED",
+            # Vault
+            "VaultCreated": "VAULT_CREATED",
+            "Deposited": "VAULT_DEPOSITED",
+            "Withdrawn": "VAULT_WITHDRAWN"
+        }
+
+        normalized_type = type_mapping.get(event_name, f"EVM_{event_name.upper()}")
+
+        return NormalizedEvent(
+            chain_id=self.chain_id,
+            block_number=block_number,
+            tx_hash=tx_hash,
+            log_index=log_index,
+            event_type=normalized_type,
+            contract_address=contract_addr,
+            payload=payload,
+            timestamp=block_timestamp
+        )

src/ingestion/fetcher.py

@@ -0,0 +1,50 @@
+import asyncio
+import logging
+from typing import List, Dict, Any, Optional
+from web3 import AsyncWeb3, AsyncHTTPProvider
+from web3.types import LogReceipt, BlockNumber
+from src.config import indexer_settings
+
+logger = logging.getLogger(__name__)
+
+class EVMFetcher:
+    """EVM 批量日志与区块抓取器 (Layer 1: Ingestion)"""
+
+    def __init__(self, rpc_url: Optional[str] = None, chain_id: Optional[int] = None):
+        self.rpc_url = rpc_url or indexer_settings.RPC_URL
+        self.chain_id = chain_id or indexer_settings.CHAIN_ID
+        self.w3 = AsyncWeb3(AsyncHTTPProvider(self.rpc_url))
+
+    async def get_latest_block_number(self) -> int:
+        """获取链上最新安全区块高度"""
+        block_num = await self.w3.eth.block_number
+        return max(0, block_num - indexer_settings.CONFIRMATIONS)
+
+    async def fetch_logs(
+        self,
+        from_block: int,
+        to_block: int,
+        addresses: Optional[List[str]] = None,
+        topics: Optional[List[Any]] = None
+    ) -> List[LogReceipt]:
+        """批量获取指定区块区间的日志"""
+        filter_params: Dict[str, Any] = {
+            "fromBlock": from_block,
+            "toBlock": to_block
+        }
+        if addresses:
+            filter_params["address"] = [self.w3.to_checksum_address(a) for a in addresses]
+        if topics:
+            filter_params["topics"] = topics
+
+        try:
+            logs = await self.w3.eth.get_logs(filter_params)
+            return logs
+        except Exception as e:
+            logger.error(f"Error fetching logs from {from_block} to {to_block}: {e}")
+            raise e
+
+    async def get_block_timestamp(self, block_number: int) -> int:
+        """获取区块出块时间戳"""
+        block = await self.w3.eth.get_block(block_number)
+        return block.get("timestamp", 0)

src/main.py

@@ -0,0 +1,122 @@
+import asyncio
+import logging
+import signal
+import sys
+from typing import Optional, Set
+import sys
+import os
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+from src.config import indexer_settings
+from src.ingestion.fetcher import EVMFetcher
+from src.decoding.evm import EVMDecoder
+from src.projections.store import ProjectionStore
+from src.projections.repository import KlineRepository
+from src.core.checkpoint import CheckpointState
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")
+logger = logging.getLogger("WTFX-Indexer")
+
+class IndexerService:
+    """WTFX 确定性索引服务核心执行引擎"""
+
+    def __init__(self):
+        self.fetcher = EVMFetcher(indexer_settings.RPC_URL, indexer_settings.CHAIN_ID)
+        self.decoder = EVMDecoder(indexer_settings.CHAIN_ID)
+        self.projection_store = ProjectionStore()
+        self.checkpoint = CheckpointState.load(
+            indexer_settings.CHECKPOINT_FILE,
+            indexer_settings.CHAIN_ID,
+            indexer_settings.START_BLOCK
+        )
+        self.repository = KlineRepository(
+            indexer_settings.DATABASE_URL.replace("postgresql+asyncpg://", "postgresql://", 1)
+        )
+        # 动态关注地址：控制器 + 金库工厂 + 已知默认市场；新市场由 CreateNewMarket 动态加入
+        self.watch_addresses: Set[str] = set()
+        self.is_running = False
+
+    async def _bootstrap_watch_list(self):
+        self.watch_addresses.add(indexer_settings.CONTROLLER_ADDRESS.lower())
+        self.watch_addresses.add(indexer_settings.VAULT_FACTORY_ADDRESS.lower())
+        default_market = os.getenv("DEFAULT_MARKET", "0x048E9a90C25ba2c4410425D282b19A472e076039")
+        self.watch_addresses.add(default_market.lower())
+        logger.info(f"Watch list initialized: {sorted(self.watch_addresses)}")
+
+    async def run(self):
+        self.is_running = True
+        await self.repository.connect()
+        await self._bootstrap_watch_list()
+        logger.info(f"Starting WTFX Indexer on Chain {indexer_settings.CHAIN_ID} (RPC: {indexer_settings.RPC_URL})")
+        logger.info(f"Resuming from block {self.checkpoint.last_processed_block}")
+
+        while self.is_running:
+            try:
+                latest_block = await self.fetcher.get_latest_block_number()
+                current_block = self.checkpoint.last_processed_block
+
+                if current_block < latest_block:
+                    to_block = min(current_block + indexer_settings.BATCH_SIZE, latest_block)
+                    logger.info(f"Scanning blocks {current_block + 1} -> {to_block} (Latest: {latest_block})...")
+
+                    logs = await self.fetcher.fetch_logs(
+                        from_block=current_block + 1,
+                        to_block=to_block,
+                        addresses=sorted(self.watch_addresses)
+                    )
+
+                    # 一次性抓取本批涉及的区块时间戳，保证 K 线时间桶精确
+                    block_nums = {int(log.get("blockNumber", 0)) for log in logs}
+                    timestamps = {}
+                    for bn in block_nums:
+                        timestamps[bn] = await self.fetcher.get_block_timestamp(bn)
+
+                    decoded_count = 0
+                    for log in logs:
+                        bn = int(log.get("blockNumber", 0))
+                        normalized = self.decoder.decode_log(log, block_timestamp=timestamps.get(bn, 0))
+                        if normalized:
+                            decoded_count += 1
+                            if normalized.event_type == "MARKET_DEPLOYED":
+                                market_addr = (normalized.payload.get("market") or "").lower()
+                                if market_addr:
+                                    self.watch_addresses.add(market_addr)
+                                    logger.info(f"New market watched: {market_addr}")
+                            self.projection_store.apply_event(normalized)
+
+                    # 持久化脏数据 (trades + klines)
+                    await self.repository.flush(
+                        self.projection_store.get_dirty_trades(),
+                        self.projection_store.get_dirty_klines()
+                    )
+                    self.projection_store.mark_trades_flushed()
+                    self.projection_store.mark_klines_flushed()
+
+                    # 推进并持久化 checkpoint
+                    self.checkpoint.last_processed_block = to_block
+                    self.checkpoint.updated_at = int(asyncio.get_event_loop().time())
+                    self.checkpoint.save(indexer_settings.CHECKPOINT_FILE)
+                    logger.info(f"Batch done. Decoded {decoded_count} events. Checkpoint -> {to_block}")
+                else:
+                    await asyncio.sleep(indexer_settings.POLL_INTERVAL_SECONDS)
+
+            except Exception as e:
+                logger.error(f"Error in indexer polling loop: {e}", exc_info=True)
+                await asyncio.sleep(indexer_settings.POLL_INTERVAL_SECONDS * 2)
+
+    def stop(self):
+        logger.info("Stopping Indexer service...")
+        self.is_running = False
+
+async def main():
+    service = IndexerService()
+    loop = asyncio.get_running_loop()
+    for sig in (signal.SIGINT, signal.SIGTERM):
+        try:
+            loop.add_signal_handler(sig, service.stop)
+        except NotImplementedError:
+            pass  # Windows compatibility
+    await service.run()
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/projections/repository.py

@@ -0,0 +1,127 @@
+import logging
+from typing import Any, Dict, List, Tuple
+
+import asyncpg
+
+logger = logging.getLogger(__name__)
+
+CREATE_TABLES_SQL = """
+CREATE TABLE IF NOT EXISTS klines (
+    market_address VARCHAR(42) NOT NULL,
+    outcome_index  INTEGER NOT NULL,
+    bar_time       BIGINT NOT NULL,
+    open           NUMERIC NOT NULL,
+    high           NUMERIC NOT NULL,
+    low            NUMERIC NOT NULL,
+    close          NUMERIC NOT NULL,
+    volume         NUMERIC NOT NULL,
+    PRIMARY KEY (market_address, outcome_index, bar_time)
+);
+CREATE INDEX IF NOT EXISTS idx_klines_market_time
+    ON klines (market_address, outcome_index, bar_time);
+
+CREATE TABLE IF NOT EXISTS trades (
+    id             BIGSERIAL PRIMARY KEY,
+    tx_hash        VARCHAR(66),
+    market_address VARCHAR(42) NOT NULL,
+    outcome_index  INTEGER NOT NULL,
+    trade_type     VARCHAR(8) NOT NULL,
+    price          NUMERIC NOT NULL,
+    volume         NUMERIC NOT NULL,
+    block_number   BIGINT,
+    ts             BIGINT
+);
+CREATE INDEX IF NOT EXISTS idx_trades_market_ts
+    ON trades (market_address, outcome_index, ts);
+"""
+
+UPSERT_KLINE_SQL = """
+INSERT INTO klines (market_address, outcome_index, bar_time, open, high, low, close, volume)
+VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
+ON CONFLICT (market_address, outcome_index, bar_time)
+DO UPDATE SET
+    high   = GREATEST(klines.high, EXCLUDED.high),
+    low    = LEAST(klines.low, EXCLUDED.low),
+    close  = EXCLUDED.close,
+    volume = klines.volume + EXCLUDED.volume;
+"""
+
+INSERT_TRADE_SQL = """
+INSERT INTO trades (tx_hash, market_address, outcome_index, trade_type, price, volume, block_number, ts)
+VALUES ($1, $2, $3, $4, $5, $6, $7, $8);
+"""
+
+SELECT_KLINES_SQL = """
+SELECT bar_time, open, high, low, close, volume
+FROM klines
+WHERE market_address = $1 AND outcome_index = $2
+ORDER BY bar_time ASC
+LIMIT $3;
+"""
+
+
+class KlineRepository:
+    """indexer -> PostgreSQL 的真实 K 线 / 成交持久化"""
+
+    def __init__(self, database_url: str):
+        self.database_url = database_url
+        self._pool: asyncpg.Pool | None = None
+
+    async def connect(self):
+        self._pool = await asyncpg.create_pool(dsn=self.database_url, min_size=1, max_size=3)
+        async with self._pool.acquire() as conn:
+            await conn.execute(CREATE_TABLES_SQL)
+        logger.info("KlineRepository connected to PostgreSQL, tables ensured.")
+
+    async def close(self):
+        if self._pool:
+            await self._pool.close()
+            self._pool = None
+
+    async def flush(self, trades: list, klines: Dict[Tuple[str, int], List[Dict[str, Any]]]):
+        if not self._pool:
+            return
+        async with self._pool.acquire() as conn:
+            async with conn.transaction():
+                for t in trades:
+                    await conn.execute(
+                        INSERT_TRADE_SQL,
+                        t.get("tx_hash"),
+                        t.get("market"),
+                        t.get("outcome_index", 0),
+                        t.get("type", "MINT"),
+                        t.get("price", 0),
+                        t.get("collateral_amount", 0),
+                        t.get("block_number"),
+                        t.get("timestamp", 0),
+                    )
+                for (m_addr, o_idx), bars in klines.items():
+                    for bar in bars:
+                        await conn.execute(
+                            UPSERT_KLINE_SQL,
+                            m_addr,
+                            o_idx,
+                            bar["time"],
+                            bar["open"],
+                            bar["high"],
+                            bar["low"],
+                            bar["close"],
+                            bar["volume"],
+                        )
+
+    async def get_klines(self, market_address: str, outcome_index: int = 0, limit: int = 1000):
+        if not self._pool:
+            return []
+        async with self._pool.acquire() as conn:
+            rows = await conn.fetch(SELECT_KLINES_SQL, market_address.lower(), outcome_index, limit)
+        return [
+            {
+                "time": r["bar_time"],
+                "open": float(r["open"]),
+                "high": float(r["high"]),
+                "low": float(r["low"]),
+                "close": float(r["close"]),
+                "volume": float(r["volume"]),
+            }
+            for r in rows
+        ]

src/projections/store.py

@@ -0,0 +1,185 @@
+import logging
+import math
+from typing import Dict, Any, List, Optional
+from src.core.events import NormalizedEvent
+
+logger = logging.getLogger(__name__)
+
+def _to_int(value: Any, default: int = 0) -> int:
+    """robust int parsing: int | '0x..' | '000..1' (indexed topic hex) | '12'"""
+    if value is None:
+        return default
+    if isinstance(value, int):
+        return value
+    if isinstance(value, str):
+        s = value[2:] if value.startswith("0x") else value
+        try:
+            return int(s, 16)
+        except ValueError:
+            try:
+                return int(s)
+            except ValueError:
+                return default
+    return default
+
+class ProjectionStore:
+    """
+    【确定性业务投影与秒级 OHLCV K 线聚合引擎】
+    - 记录逐笔成交 (Trades)
+    - 动态维护市场各 Token 瞬时价格与流动性
+    - 按照 5s 粒度聚合实时 OHLCV K 线（适配 TradingView 图表）
+    - 内存投影 + 脏数据队列，由 main 定期 flush 到 PostgreSQL
+    """
+
+    def __init__(self):
+        self.markets: Dict[str, Dict[str, Any]] = {}
+        self.trades: list = []
+        self.positions: Dict[str, Dict[str, Any]] = {}
+        self.processed_events: set = set()
+
+        # K 线存储结构: market_address -> outcome_index -> List[OHLCV]
+        # item: {"time": 1700000000, "open": 0.5, "high": 0.52, "low": 0.48, "close": 0.51, "volume": 1200}
+        self.klines: Dict[str, Dict[int, List[Dict[str, Any]]]] = {}
+
+        # 脏数据追踪（供 flush 到 DB 使用）
+        self._trade_watermark = 0
+        self._dirty_klines: Dict[tuple, List[Dict[str, Any]]] = {}
+
+    def reset(self):
+        self.markets.clear()
+        self.trades.clear()
+        self.positions.clear()
+        self.processed_events.clear()
+        self.klines.clear()
+        self._trade_watermark = 0
+        self._dirty_klines.clear()
+        logger.info("Projection & K-line store reset successfully.")
+
+    def get_klines(self, market_address: str, outcome_index: int = 0) -> List[Dict[str, Any]]:
+        m_addr = market_address.lower()
+        if m_addr in self.klines and outcome_index in self.klines[m_addr]:
+            return self.klines[m_addr][outcome_index]
+        return []
+
+    def get_dirty_trades(self) -> list:
+        """自上次 flush 以来的新增成交"""
+        new_trades = self.trades[self._trade_watermark:]
+        return new_trades
+
+    def mark_trades_flushed(self):
+        self._trade_watermark = len(self.trades)
+
+    def get_dirty_klines(self) -> Dict[tuple, List[Dict[str, Any]]]:
+        """自上次 flush 以来被更新的 (market, outcome) -> bars"""
+        return {k: v for k, v in self._dirty_klines.items()}
+
+    def mark_klines_flushed(self):
+        self._dirty_klines.clear()
+
+    def _update_kline(self, market_address: str, outcome_index: int, price: float, volume: float, timestamp: int):
+        m_addr = market_address.lower()
+        if m_addr not in self.klines:
+            self.klines[m_addr] = {}
+        if outcome_index not in self.klines[m_addr]:
+            self.klines[m_addr][outcome_index] = []
+
+        bar_time = (timestamp // 5) * 5  # 5秒一根 K 线 Bar
+        bars = self.klines[m_addr][outcome_index]
+        key = (m_addr, outcome_index)
+
+        if bars and bars[-1]["time"] == bar_time:
+            # 更新当前柱子
+            current = bars[-1]
+            current["high"] = max(current["high"], price)
+            current["low"] = min(current["low"], price)
+            current["close"] = price
+            current["volume"] += volume
+        else:
+            # 开新柱子
+            prev_close = bars[-1]["close"] if bars else price
+            bars.append({
+                "time": bar_time,
+                "open": prev_close,
+                "high": max(prev_close, price),
+                "low": min(prev_close, price),
+                "close": price,
+                "volume": volume
+            })
+
+        self._dirty_klines[key] = bars
+
+    def apply_event(self, event: NormalizedEvent):
+        if event.event_id in self.processed_events:
+            return
+        self.processed_events.add(event.event_id)
+
+        etype = event.event_type
+        payload = event.payload
+
+        if etype == "MARKET_DEPLOYED":
+            market_addr = (payload.get("market") or event.contract_address).lower()
+            self.markets[market_addr] = {
+                "market_address": market_addr,
+                "question_id": payload.get("questionId"),
+                "curve": payload.get("curve"),
+                "collateral": payload.get("collateral"),
+                "creator": payload.get("creator"),
+                "tier": payload.get("tier", 1),
+                "fee_rate": payload.get("feeRate"),
+                "status": "ACTIVE",
+                "winning_outcome": None,
+                "created_at_block": event.block_number,
+                "created_at_tx": event.tx_hash,
+                "timestamp": event.timestamp
+            }
+
+        elif etype in ("ORDER_MINT", "ORDER_REDEEM"):
+            market_addr = event.contract_address.lower()
+
+            token_id = _to_int(payload.get("tokenId")) or _to_int(payload.get("id"))
+            outcome_idx = 0
+            if token_id > 0 and (token_id & (token_id - 1)) == 0:
+                # tokenId = 2^idx -> idx
+                outcome_idx = int(math.log2(token_id))
+            else:
+                outcome_idx = _to_int(payload.get("outcomeIndex"))
+
+            if etype == "ORDER_MINT":
+                # MintSwapV2(caller, receiver, tokenId, collateralToPool, otToUser, collateralToTreasury)
+                collateral_in = _to_int(payload.get("collateralToPool")) + _to_int(payload.get("collateralToTreasury"))
+                ot_moved = _to_int(payload.get("otToUser"))
+                volume_raw = collateral_in
+            else:
+                # RedeemSwapV2(caller, receiver, tokenId, collateralFromPool, otToPool, collateralToTreasury)
+                collateral_out = _to_int(payload.get("collateralFromPool")) + _to_int(payload.get("collateralToTreasury"))
+                ot_moved = _to_int(payload.get("otToPool"))
+                volume_raw = collateral_out
+
+            collateral_raw = volume_raw / 1e18
+            tokens_raw = ot_moved / 1e18
+
+            # 成交均价 = 移动的抵押品 / 移动的 OT。PowerLDA 价格可 > 1，不做概率区间裁剪
+            trade_price = collateral_raw / tokens_raw if tokens_raw > 0 else 0.0
+
+            self.trades.append({
+                "tx_hash": event.tx_hash,
+                "block_number": event.block_number,
+                "market": market_addr,
+                "type": "MINT" if etype == "ORDER_MINT" else "REDEEM",
+                "user": payload.get("receiver") or payload.get("user"),
+                "outcome_index": outcome_idx,
+                "collateral_amount": collateral_raw,
+                "tokens_amount": tokens_raw,
+                "price": trade_price,
+                "fee": _to_int(payload.get("collateralToTreasury")) / 1e18,
+                "timestamp": event.timestamp or int(event.block_number)
+            })
+
+            # 更新 OHLCV K 线
+            self._update_kline(
+                market_address=market_addr,
+                outcome_index=outcome_idx,
+                price=trade_price,
+                volume=collateral_raw,
+                timestamp=event.timestamp or int(event.block_number)
+            )

tests/test_indexer_pipeline.py

@@ -0,0 +1,76 @@
+import pytest
+from src.core.events import NormalizedEvent
+from src.projections.store import ProjectionStore
+from src.decoding.evm import EVMDecoder
+
+def test_normalized_event_idempotency():
+    event = NormalizedEvent(
+        chain_id=46630,
+        block_number=1000,
+        tx_hash="0xabcdef123456",
+        log_index=2,
+        event_type="MARKET_DEPLOYED",
+        contract_address="0xc0E24E152771C588B21AEB654b30B1cBAf381c1a",
+        payload={
+            "market": "0x1111222233334444555566667777888899990000",
+            "questionId": "0x9999",
+            "tier": 1,
+            "creator": "0x6cddF384792C77219fc25C454Cfe264757842830"
+        },
+        timestamp=1700000000
+    )
+    assert event.event_id == "46630:0xabcdef123456:2"
+
+def test_projection_store_deploy_and_resolve():
+    store = ProjectionStore()
+
+    deploy_event = NormalizedEvent(
+        chain_id=46630,
+        block_number=1000,
+        tx_hash="0xabcdef123456",
+        log_index=0,
+        event_type="MARKET_DEPLOYED",
+        contract_address="0xc0E24E152771C588B21AEB654b30B1cBAf381c1a",
+        payload={
+            "market": "0x1111222233334444555566667777888899990000",
+            "questionId": "0x9999",
+            "tier": 1,
+            "creator": "0x6cddF384792C77219fc25C454Cfe264757842830"
+        },
+        timestamp=1700000000
+    )
+    store.apply_event(deploy_event)
+    assert "0x1111222233334444555566667777888899990000" in store.markets
+    assert store.markets["0x1111222233334444555566667777888899990000"]["status"] == "ACTIVE"
+
+    # Duplicate should be ignored
+    store.apply_event(deploy_event)
+    assert len(store.markets) == 1
+
+    # Resolve event
+    resolve_event = NormalizedEvent(
+        chain_id=46630,
+        block_number=1050,
+        tx_hash="0xabcdef789012",
+        log_index=1,
+        event_type="OUTCOME_RESOLVED",
+        contract_address="0xc0E24E152771C588B21AEB654b30B1cBAf381c1a",
+        payload={
+            "questionId": "0x9999",
+            "outcome": 1
+        },
+        timestamp=1700005000
+    )
+    store.apply_event(resolve_event)
+    assert store.markets["0x1111222233334444555566667777888899990000"]["status"] == "RESOLVED"
+    assert store.markets["0x1111222233334444555566667777888899990000"]["tentative_winning_outcome"] == 1
+
+    # Test Reset & Replay
+    store.reset()
+    assert len(store.markets) == 0
+    assert len(store.processed_events) == 0
+
+    # Replay
+    store.apply_event(deploy_event)
+    store.apply_event(resolve_event)
+    assert store.markets["0x1111222233334444555566667777888899990000"]["status"] == "RESOLVED"