Refactor sql models

CLAUDE.md

@@ -232,6 +232,54 @@ dbt docs generate
 dbt docs serve
 ```
 
+### Testing
+```bash
+# Run all tests
+uv run pytest
+
+# Run specific test file
+uv run pytest tests/test_connection.py
+
+# Run with verbose output
+uv run pytest -v
+```
+
+### Linting (dbt SQL)
+```bash
+# Lint dbt models with SQLFluff
+uv run sqlfluff lint dbt_project/models/
+
+# Fix auto-fixable issues
+uv run sqlfluff fix dbt_project/models/
+```
+
+### Kafka Streaming (Speed Layer)
+The project includes a Kafka-based streaming layer for near real-time event processing.
+
+**CLI Entry Points** (defined in pyproject.toml):
+```bash
+# Initialize Kafka schema/topics
+uv run kafka-init-schema
+
+# Start Kafka producer (streams events from HyperSync)
+uv run kafka-producer
+
+# Start Kafka consumer (writes to database)
+uv run kafka-consumer
+```
+
+**Direct Script Execution:**
+```bash
+# Producer - streams blockchain events to Kafka
+uv run python scripts/kafka/producer.py
+
+# Consumer - consumes from Kafka and loads to database
+uv run python scripts/kafka/consumer.py
+
+# Batch consumer (for Airflow integration)
+uv run python scripts/kafka/consume_batch.py
+```
+
 ## Environment Variables
 
 Required variables (see [.env.example](.env.example)):

chat_engine/chat_engine.py

@@ -80,17 +80,6 @@ def chat_engine(query: str, enabled_agents: Optional[List[str]] = None):
     return graph.invoke(state)
 
 
-def visualize_graph():
-    """Display the graph structure (for Jupyter notebooks)."""
-    try:
-        from IPython.display import Image, display
-
-        graph = build_graph()
-        display(Image(graph.get_graph().draw_png()))
-    except ImportError:
-        print("IPython not available. Graph visualization skipped.")
-
-
 def main():
     """Main execution function with example queries."""
     # query1 = "What is openai's latest product?"

dbt_project/models/debug_conversion.sql

@@ -1,6 +1,6 @@
 {{ config(materialized='view') }}
 
--- Debug model to check hex conversion
+-- Debug model to check hex conversion for Transfer events
 -- Run: dbt run --select debug_conversion --target prod
 
 with sample_data as (