mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-09-15 04:00:33 +00:00
29 KiB
29 KiB
In [ ]:
from semantica.ingest import StreamIngestor, WebIngestor, DBIngestor, FileIngestor
from semantica.parse import JSONParser, StructuredDataParser
from semantica.semantic_extract import NERExtractor, RelationExtractor, EventDetector, TripleExtractor
from semantica.kg import GraphBuilder, TemporalGraphQuery, TemporalPatternDetector, GraphAnalyzer
from semantica.kg import CentralityCalculator, CommunityDetector, ConnectivityAnalyzer
from semantica.reasoning import InferenceEngine, RuleManager, ExplanationGenerator
from semantica.kg_qa import KGQualityAssessor
from semantica.kg import ConflictDetector
from semantica.export import JSONExporter, RDFExporter, ReportGenerator
from semantica.visualization import KGVisualizer, TemporalVisualizer, AnalyticsVisualizer
import tempfile
import os
import json
from datetime import datetime, timedelta
stream_ingestor = StreamIngestor()
web_ingestor = WebIngestor()
db_ingestor = DBIngestor()
file_ingestor = FileIngestor()
json_parser = JSONParser()
structured_parser = StructuredDataParser()
# Real streaming sources for blockchain transactions
stream_sources = [
{
"type": "kafka",
"topic": "blockchain_transactions",
"bootstrap_servers": ["localhost:9092"],
"consumer_config": {"group_id": "transaction_analysis"}
},
{
"type": "rabbitmq",
"queue": "eth_transactions",
"connection_url": "amqp://user:password@localhost:5672/"
}
]
# Real blockchain APIs
blockchain_apis = [
"https://api.etherscan.io/api?module=proxy&action=eth_getBlockByNumber&tag=latest&boolean=true&apikey=YourApiKeyToken", # Etherscan API
"https://blockchain.info/rawblock/000000000019d6689c085ae165831e934ff763ae46a2a6c172b3f1b60a8ce26f", # Blockchain.com API
"https://api.coingecko.com/api/v3/coins/ethereum" # CoinGecko API
]
# Real database connection for transaction history
db_connection_string = "postgresql://user:password@localhost:5432/blockchain_db"
db_query = "SELECT tx_hash, from_address, to_address, value, timestamp, block_number FROM transactions WHERE timestamp > NOW() - INTERVAL '24 hours' ORDER BY timestamp DESC LIMIT 10000"
temp_dir = tempfile.mkdtemp()
# Sample transaction data for local ingestion
transaction_data_file = os.path.join(temp_dir, "transactions.json")
transaction_data = [
{
"tx_hash": "0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef",
"from_address": "0xabc123def456abc123def456abc123def456abc12",
"to_address": "0xdef456abc123def456abc123def456abc123def45",
"value": "1000000000000000000",
"timestamp": (datetime.now() - timedelta(hours=1)).isoformat(),
"block_number": 18500000,
"gas_used": 21000
},
{
"tx_hash": "0x2345678901bcdef2345678901bcdef2345678901bcdef2345678901bcdef23",
"from_address": "0xdef456abc123def456abc123def456abc123def45",
"to_address": "0x7890123456789012345678901234567890123456",
"value": "500000000000000000",
"timestamp": (datetime.now() - timedelta(hours=2)).isoformat(),
"block_number": 18499950,
"gas_used": 21000
},
{
"tx_hash": "0x3456789012cdef3456789012cdef3456789012cdef3456789012cdef3456",
"from_address": "0x7890123456789012345678901234567890123456",
"to_address": "0xabc123def456abc123def456abc123def456abc12",
"value": "2000000000000000000",
"timestamp": (datetime.now() - timedelta(hours=3)).isoformat(),
"block_number": 18499900,
"gas_used": 21000
},
{
"tx_hash": "0x4567890123def4567890123def4567890123def4567890123def4567890123",
"from_address": "0xabc123def456abc123def456abc123def456abc12",
"to_address": "0x4567890123456789012345678901234567890123",
"value": "300000000000000000",
"timestamp": (datetime.now() - timedelta(hours=4)).isoformat(),
"block_number": 18499850,
"gas_used": 21000
},
{
"tx_hash": "0x5678901234ef5678901234ef5678901234ef5678901234ef5678901234ef56",
"from_address": "0x4567890123456789012345678901234567890123",
"to_address": "0x1234567890123456789012345678901234567890",
"value": "1500000000000000000",
"timestamp": (datetime.now() - timedelta(hours=5)).isoformat(),
"block_number": 18499800,
"gas_used": 21000
}
]
with open(transaction_data_file, 'w') as f:
json.dump(transaction_data, f, indent=2)
# Ingest from local file
file_data = file_ingestor.ingest_file(transaction_data_file)
parsed_transactions = structured_parser.parse_json(json.dumps(transaction_data))
# Ingest from blockchain APIs (example with public API)
try:
web_content = web_ingestor.ingest_url(blockchain_apis[2]) # CoinGecko public API
if web_content:
print(f"✓ Ingested web content: {web_content.url if hasattr(web_content, 'url') else 'N/A'}")
except Exception as e:
print(f"⚠ Web ingestion (example): {str(e)[:100]}")
# Database ingestion pattern (would connect to real database)
try:
db_data = db_ingestor.export_table(
connection_string=db_connection_string,
table_name="transactions",
limit=10000
)
print(f"✓ Database ingestion configured for: {db_connection_string}")
print(f" Query pattern: {db_query}")
except Exception as e:
print(f"⚠ Database connection (example pattern): Configure with real credentials")
db_data = {"data": transaction_data}
# Streaming ingestion pattern
print(f"✓ Streaming sources configured:")
for stream_source in stream_sources:
print(f" - {stream_source['type']}: {stream_source.get('topic') or stream_source.get('queue')}")
print(f"\n📊 Ingestion Summary:")
print(f" Local transactions: {len(transaction_data)}")
print(f" Database records: {len(db_data.get('data', [])) if db_data else 0}")
print(f" Streaming sources: {len(stream_sources)}")
print(f" Web APIs: {len(blockchain_apis)}")
In [ ]:
ner_extractor = NERExtractor()
relation_extractor = RelationExtractor()
event_detector = EventDetector()
triple_extractor = TripleExtractor()
all_transaction_texts = []
all_transactions = []
# Process parsed transactions
if parsed_transactions and isinstance(parsed_transactions, dict):
transactions = parsed_transactions.get("data", transaction_data)
for tx in transactions:
all_transactions.append(tx)
tx_text = f"Transaction {tx.get('tx_hash', '')} from {tx.get('from_address', '')} to {tx.get('to_address', '')} value {tx.get('value', '')} at {tx.get('timestamp', '')}"
all_transaction_texts.append(tx_text)
# Extract entities
all_entities = []
all_relationships = []
all_events = []
all_triples = []
for text in all_transaction_texts:
entities = ner_extractor.extract(text)
all_entities.extend(entities)
relationships = relation_extractor.extract(text, entities)
all_relationships.extend(relationships)
events = event_detector.detect_events(text)
all_events.extend(events)
triples = triple_extractor.extract(text)
all_triples.extend(triples)
# Build structured entity list
transaction_entities = []
wallet_entities = []
for tx in all_transactions:
tx_entity = {
"id": tx.get("tx_hash", ""),
"type": "Transaction",
"properties": {
"from_address": tx.get("from_address", ""),
"to_address": tx.get("to_address", ""),
"value": tx.get("value", ""),
"timestamp": tx.get("timestamp", ""),
"block_number": tx.get("block_number", 0),
"gas_used": tx.get("gas_used", 0)
}
}
transaction_entities.append(tx_entity)
# Add wallet entities
from_wallet = {
"id": tx.get("from_address", ""),
"type": "Wallet",
"properties": {
"address": tx.get("from_address", ""),
"role": "sender"
}
}
to_wallet = {
"id": tx.get("to_address", ""),
"type": "Wallet",
"properties": {
"address": tx.get("to_address", ""),
"role": "receiver"
}
}
wallet_entities.append(from_wallet)
wallet_entities.append(to_wallet)
# Deduplicate wallets
unique_wallets = {}
for wallet in wallet_entities:
wallet_id = wallet["id"]
if wallet_id not in unique_wallets:
unique_wallets[wallet_id] = wallet
wallet_entities = list(unique_wallets.values())
print(f"Extracted {len(transaction_entities)} transactions")
print(f"Extracted {len(wallet_entities)} unique wallets")
print(f"Extracted {len(all_relationships)} relationships")
print(f"Detected {len(all_events)} events")
print(f"Extracted {len(all_triples)} triples")
In [ ]:
builder = GraphBuilder()
# Add all entities
for wallet in wallet_entities:
builder.add_entity(
entity_id=wallet["id"],
entity_type=wallet["type"],
properties=wallet.get("properties", {})
)
for tx in transaction_entities:
builder.add_entity(
entity_id=tx["id"],
entity_type=tx["type"],
properties=tx.get("properties", {})
)
# Add relationships
relationships = []
for tx in transaction_entities:
from_addr = tx["properties"].get("from_address", "")
to_addr = tx["properties"].get("to_address", "")
tx_hash = tx["id"]
value = tx["properties"].get("value", "")
timestamp = tx["properties"].get("timestamp", "")
# Transaction relationship
rel = {
"source": from_addr,
"target": to_addr,
"type": "transfers_to",
"properties": {
"transaction": tx_hash,
"value": value,
"timestamp": timestamp
}
}
relationships.append(rel)
builder.add_relationship(
source_id=from_addr,
target_id=to_addr,
relationship_type="transfers_to",
properties=rel["properties"]
)
# Transaction entity relationship
builder.add_relationship(
source_id=from_addr,
target_id=tx_hash,
relationship_type="initiates",
properties={"timestamp": timestamp}
)
builder.add_relationship(
source_id=tx_hash,
target_id=to_addr,
relationship_type="sends_to",
properties={"timestamp": timestamp}
)
knowledge_graph = builder.build()
print(f"Built knowledge graph with {len(knowledge_graph.nodes)} nodes")
print(f"Built knowledge graph with {len(knowledge_graph.edges)} edges")
print(f"Added {len(relationships)} transaction relationships")
In [ ]:
temporal_query = TemporalGraphQuery(knowledge_graph)
pattern_detector = TemporalPatternDetector(knowledge_graph)
graph_analyzer = GraphAnalyzer(knowledge_graph)
centrality_calculator = CentralityCalculator(knowledge_graph)
community_detector = CommunityDetector(knowledge_graph)
connectivity_analyzer = ConnectivityAnalyzer(knowledge_graph)
# Query transactions in time range
start_time = (datetime.now() - timedelta(hours=6)).isoformat()
end_time = datetime.now().isoformat()
temporal_results = temporal_query.query_time_range(
start_time=start_time,
end_time=end_time,
relationship_types=["transfers_to", "initiates", "sends_to"]
)
# Detect temporal patterns
temporal_patterns = pattern_detector.detect_temporal_patterns(
relationship_types=["transfers_to"],
time_window_hours=6
)
# Calculate centrality to find key wallets
centrality_scores = centrality_calculator.calculate_centrality(centrality_type="betweenness")
top_central_wallets = sorted(centrality_scores.items(), key=lambda x: x[1], reverse=True)[:10]
# Detect communities (clustering)
communities = community_detector.detect_communities()
community_count = len(set(communities.values())) if communities else 0
# Analyze connectivity
connectivity_results = connectivity_analyzer.analyze_connectivity()
# AML Pattern Detection using Inference Engine
inference_engine = InferenceEngine()
rule_manager = RuleManager()
# Define AML rules
aml_rules = [
{
"name": "tumbling_pattern",
"condition": "high_transaction_count AND multiple_intermediate_wallets",
"action": "flag_as_tumbling"
},
{
"name": "mixing_pattern",
"condition": "funds_split_into_multiple_addresses AND rapid_consolidation",
"action": "flag_as_mixing"
},
{
"name": "suspicious_flow",
"condition": "large_value_transfer AND short_time_window",
"action": "flag_as_suspicious"
}
]
for rule in aml_rules:
rule_manager.add_rule(rule["name"], rule["condition"], rule["action"])
# Add facts from graph analysis
aml_facts = []
for wallet_id, centrality in top_central_wallets[:5]:
aml_facts.append({
"wallet": wallet_id,
"centrality": centrality,
"high_transaction_count": True if centrality > 0.1 else False
})
# Detect patterns
suspicious_patterns = []
for wallet_id, centrality in top_central_wallets:
if centrality > 0.15:
suspicious_patterns.append({
"wallet": wallet_id,
"pattern": "high_centrality",
"risk_score": min(centrality * 10, 10),
"description": f"Wallet {wallet_id[:10]}... has high betweenness centrality ({centrality:.3f}), indicating potential mixing/tumbling"
})
# Check for rapid transactions (tumbling pattern)
wallet_transaction_counts = {}
for rel in relationships:
source = rel["source"]
wallet_transaction_counts[source] = wallet_transaction_counts.get(source, 0) + 1
for wallet_id, count in wallet_transaction_counts.items():
if count >= 3:
suspicious_patterns.append({
"wallet": wallet_id,
"pattern": "rapid_transactions",
"risk_score": min(count * 2, 10),
"description": f"Wallet {wallet_id[:10]}... has {count} outgoing transactions, potential tumbling"
})
print(f"Detected {len(temporal_patterns)} temporal patterns")
print(f"Found {community_count} wallet communities")
print(f"Identified {len(suspicious_patterns)} suspicious patterns")
print(f"\nTop 5 Central Wallets:")
for i, (wallet_id, centrality) in enumerate(top_central_wallets[:5], 1):
print(f" {i}. {wallet_id[:20]}... (centrality: {centrality:.3f})")
print(f"\nSuspicious Patterns Detected:")
for pattern in suspicious_patterns[:5]:
print(f" - {pattern['pattern']}: {pattern['wallet'][:20]}... (risk: {pattern['risk_score']:.1f}/10)")
In [ ]:
json_exporter = JSONExporter()
rdf_exporter = RDFExporter()
report_generator = ReportGenerator()
kg_quality_assessor = KGQualityAssessor()
conflict_detector = ConflictDetector(knowledge_graph)
# Assess graph quality
quality_metrics = kg_quality_assessor.assess_quality(knowledge_graph)
# Detect conflicts
conflicts = conflict_detector.detect_conflicts()
# Generate alerts
alerts = []
for pattern in suspicious_patterns:
if pattern["risk_score"] >= 5.0:
alerts.append({
"alert_id": f"AML_{pattern['wallet'][:8]}",
"type": "AML_SUSPICIOUS_PATTERN",
"severity": "HIGH" if pattern["risk_score"] >= 7.0 else "MEDIUM",
"wallet": pattern["wallet"],
"pattern": pattern["pattern"],
"risk_score": pattern["risk_score"],
"description": pattern["description"],
"timestamp": datetime.now().isoformat()
})
# Export knowledge graph
kg_json = json_exporter.export(knowledge_graph, output_path=os.path.join(temp_dir, "transaction_kg.json"))
kg_rdf = rdf_exporter.export(knowledge_graph, output_path=os.path.join(temp_dir, "transaction_kg.rdf"))
# Generate report
report_content = f"""
# Blockchain Transaction Network Analysis Report
## Executive Summary
- Total Transactions Analyzed: {len(transaction_entities)}
- Unique Wallets: {len(wallet_entities)}
- Suspicious Patterns Detected: {len(suspicious_patterns)}
- High-Risk Alerts: {len([a for a in alerts if a['severity'] == 'HIGH'])}
## Graph Quality Metrics
- Nodes: {quality_metrics.get('node_count', len(knowledge_graph.nodes))}
- Edges: {quality_metrics.get('edge_count', len(knowledge_graph.edges))}
- Completeness: {quality_metrics.get('completeness', 0):.2%}
- Consistency: {quality_metrics.get('consistency', 0):.2%}
## Top Suspicious Patterns
"""
for i, pattern in enumerate(suspicious_patterns[:10], 1):
report_content += f"""
### {i}. {pattern['pattern'].upper()}
- Wallet: {pattern['wallet']}
- Risk Score: {pattern['risk_score']:.1f}/10
- Description: {pattern['description']}
"""
report_content += f"""
## Alerts Generated
"""
for alert in alerts:
report_content += f"""
- **{alert['alert_id']}** ({alert['severity']}): {alert['description']}
"""
report_path = os.path.join(temp_dir, "aml_analysis_report.md")
with open(report_path, 'w') as f:
f.write(report_content)
print(f"Generated {len(alerts)} AML alerts")
print(f"Exported knowledge graph to JSON and RDF")
print(f"Generated analysis report: {report_path}")
print(f"\nQuality Metrics:")
print(f" Nodes: {quality_metrics.get('node_count', len(knowledge_graph.nodes))}")
print(f" Edges: {quality_metrics.get('edge_count', len(knowledge_graph.edges))}")
print(f" Completeness: {quality_metrics.get('completeness', 0):.2%}")
print(f" Consistency: {quality_metrics.get('consistency', 0):.2%}")
In [ ]:
kg_visualizer = KGVisualizer()
temporal_visualizer = TemporalVisualizer()
analytics_visualizer = AnalyticsVisualizer()
# Visualize knowledge graph
kg_viz = kg_visualizer.visualize(
knowledge_graph,
layout="force_directed",
highlight_nodes=[p["wallet"] for p in suspicious_patterns[:5]],
node_size_by="centrality"
)
# Visualize temporal patterns
temporal_viz = temporal_visualizer.visualize(
knowledge_graph,
time_attribute="timestamp",
relationship_types=["transfers_to"]
)
# Visualize analytics
analytics_viz = analytics_visualizer.visualize(
knowledge_graph,
metrics={
"centrality": dict(top_central_wallets[:10]),
"communities": communities,
"connectivity": connectivity_results
}
)
print("Generated visualizations:")
print(" - Knowledge Graph: Transaction network with highlighted suspicious wallets")
print(" - Temporal Visualization: Transaction flows over time")
print(" - Analytics Visualization: Centrality, communities, and connectivity metrics")