feat: Monitor emission data

alembic/versions/2026_06_29_emission_snapshots.py

@@ -0,0 +1,47 @@
+"""Add emission_snapshots table
+
+Revision ID: f8a9b0c1d2e3
+Revises: e7f3a1b2c905
+Create Date: 2026-06-29 00:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "f8a9b0c1d2e3"
+down_revision: Union[str, Sequence[str], None] = "e7f3a1b2c905"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "emission_snapshots",
+        sa.Column("id", sa.BigInteger, primary_key=True, autoincrement=True),
+        sa.Column("hotkey", sa.Text, nullable=False),
+        sa.Column("agent_id", sa.UUID(), sa.ForeignKey("agents.agent_id"), nullable=False),
+        sa.Column("set_id", sa.Integer, sa.ForeignKey("competitions.set_id"), nullable=False),
+        sa.Column("block_number", sa.BigInteger, nullable=False),
+        sa.Column("emission", sa.Float, nullable=False),
+        sa.Column(
+            "created_at",
+            sa.TIMESTAMP(timezone=True),
+            nullable=False,
+            server_default=sa.text("NOW()"),
+        ),
+        sa.UniqueConstraint("hotkey", "block_number", name="uq_emission_snapshots_hotkey_block"),
+    )
+    op.create_index(
+        "idx_emission_snapshots_agent_set",
+        "emission_snapshots",
+        ["agent_id", "set_id"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("idx_emission_snapshots_agent_set", table_name="emission_snapshots")
+    op.drop_table("emission_snapshots")

db/models/emission.py

@@ -0,0 +1,23 @@
+from uuid import UUID
+
+import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import UUID as PG_UUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from db.base import Base, CreatedAtMixin
+
+
+class EmissionSnapshot(Base, CreatedAtMixin):
+    __tablename__ = "emission_snapshots"
+
+    id: Mapped[int] = mapped_column(sa.BigInteger, primary_key=True, autoincrement=True)
+    hotkey: Mapped[str] = mapped_column(sa.Text, nullable=False)
+    agent_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), sa.ForeignKey("agents.agent_id"), nullable=False)
+    set_id: Mapped[int] = mapped_column(sa.Integer, sa.ForeignKey("competitions.set_id"), nullable=False)
+    block_number: Mapped[int] = mapped_column(sa.BigInteger, nullable=False)
+    emission: Mapped[float] = mapped_column(sa.Float, nullable=False)
+
+    __table_args__ = (
+        sa.UniqueConstraint("hotkey", "block_number", name="uq_emission_snapshots_hotkey_block"),
+        sa.Index("idx_emission_snapshots_agent_set", "agent_id", "set_id"),
+    )

queries/emission.py

@@ -0,0 +1,58 @@
+from uuid import UUID
+
+import asyncpg
+
+from utils.database import DatabaseConnection, db_operation
+
+
+@db_operation
+async def insert_emission_snapshot(
+    conn: DatabaseConnection,
+    *,
+    hotkey: str,
+    agent_id: UUID,
+    set_id: int,
+    block_number: int,
+    emission: float,
+) -> None:
+    await conn.execute(
+        """
+        INSERT INTO emission_snapshots (hotkey, agent_id, set_id, block_number, emission)
+        VALUES ($1, $2, $3, $4, $5)
+        ON CONFLICT (hotkey, block_number) DO NOTHING
+        """,
+        hotkey,
+        agent_id,
+        set_id,
+        block_number,
+        emission,
+    )
+
+
+@db_operation
+async def get_total_emission_for_approved_agents(conn: DatabaseConnection, set_id: int) -> dict[UUID, float]:
+    rows = await conn.fetch(
+        """
+        SELECT agent_id, SUM(emission) AS total_emission
+        FROM emission_snapshots
+        WHERE set_id = $1
+        GROUP BY agent_id
+        """,
+        set_id,
+    )
+    return {row["agent_id"]: float(row["total_emission"]) for row in rows}
+
+
+@db_operation
+async def get_approved_hotkeys_for_set(conn: DatabaseConnection, set_id: int) -> list[asyncpg.Record]:
+    return await conn.fetch(
+        """
+        SELECT DISTINCT ON (a.miner_hotkey)
+            a.miner_hotkey, aa.agent_id
+        FROM approved_agents aa
+        JOIN agents a ON a.agent_id = aa.agent_id
+        WHERE aa.set_id = $1
+        ORDER BY a.miner_hotkey, aa.approved_at DESC
+        """,
+        set_id,
+    )

utils/bittensor.py

@@ -7,6 +7,7 @@
 import api.config as config
 
 if TYPE_CHECKING:
+    from bittensor.core.chain_data.neuron_info_lite import NeuronInfoLite
     from bittensor.core.types import BlockInfo
     from bittensor.utils.balance import Balance
 
@@ -162,6 +163,21 @@ async def get_emission(self, hotkey: str) -> float:
             return 0.0
         return float(neuron.emission)
 
+    async def get_neurons_lite(self, netuid: int) -> list["NeuronInfoLite"]:
+        """Retrieve all neurons (lite) for a given subnet."""
+        assert self._subtensor is not None, "Subtensor client is not initialized"
+        return await self._subtensor.neurons_lite(netuid=netuid)
+
+    async def get_blocks_until_next_epoch(self, netuid: int) -> int | None:
+        """Return the number of blocks until the next epoch for a given subnet."""
+        assert self._subtensor is not None, "Subtensor client is not initialized"
+        return await self._subtensor.blocks_until_next_epoch(netuid=netuid)
+
+    async def get_current_block(self) -> int:
+        """Return the current block number."""
+        assert self._subtensor is not None, "Subtensor client is not initialized"
+        return await self._subtensor.get_current_block()
+
 
 def validate_signed_timestamp(timestamp: int, signed_timestamp: str, hotkey: str) -> bool:
     try: