Report prepared for: BlueGreenCity-KI Project
Date: January 2026
Purpose: Investigate alternatives/additions to the static risk scoring pipeline
This report investigates dynamic knowledge graphs and deep temporal reasoning as potential alternatives or complementary approaches to the current static risk scoring pipeline in BlueGreenCity-KI.
Key Findings:
Dynamic Knowledge Graphs excel at representing complex relationships between entities (trees, environmental factors, maintenance events) and can evolve over time, enabling more sophisticated reasoning about tree health.
Deep Temporal Reasoning provides powerful tools for understanding how tree conditions change over time, predicting future states, and identifying patterns in historical data.
Hybrid Approach Recommended: Rather than replacing the current system, we recommend a phased integration that combines the interpretability of the current weighted scoring system with the advanced capabilities of knowledge graphs and temporal reasoning.
Immediate Value:
Implementation Complexity: Medium to High
Expected Timeline: 3-6 months for initial prototype, 6-12 months for full integration
The current BlueGreenCity-KI system uses a static weighted aggregation approach where:
risk = Σ(criterion_score_i × weight_i)Limitations of the Current Approach:
The goal is not to replace the current system entirely, but to explore methods that can:
Input Data Criteria Modules Output
───────────── ─────────────────── ──────
│ Trees │ │ Age │ │ Final Risk │
│ Climate │ ──────────▶ │ Vitality │ ──────▶ │ Score │
│ Streets │ │ Temperature │ │ (0-1) │
│ Context │ │ Structure │ │ │
└────────────┘ │ ... │ └─────────────┘
└────────────────┘
│
▼
Weight Aggregation
risk = Σ(score_i × w_i)
✅ Transparent: Easy to understand and explain
✅ Fast: Efficient computation for 50,000+ trees
✅ Modular: Easy to add new criteria
✅ Well-documented: Clear methodology
✅ Validated: Based on established literature
❌ Static: Cannot adapt to changing conditions
❌ Linear: Assumes additive effects
❌ Snapshot-based: Limited temporal understanding
❌ Isolated criteria: No interaction modeling
❌ Fixed rules: Cannot learn from data
A knowledge graph is a way of storing information that emphasizes the relationships between things rather than just the things themselves.
Simple Analogy: Think of it like a social network map:
Traditional Table (CSV/Database):
tree_id | age | vitality | street_type | temperature
12345 | 75 | 2 | motorway | 31.5
Knowledge Graph Representation:
(Tree:12345)──[HAS_VITALITY]──>(Vitality:Fair)
│
├──[HAS_AGE]──>(Age:75_years)
│
├──[LOCATED_NEAR]──>(Street:Highway_A2)
│ │
│ └──[HAS_POLLUTION_LEVEL]──>(Pollution:High)
│
└──[EXPOSED_TO]──>(Temperature:31.5C)
│
└──[CLASSIFIED_AS]──>(HeatLevel:Extreme)
A dynamic knowledge graph can:
located_in, near, within_radius_ofoccurred_after, valid_from, valid_untilcaused_by, contributes_to, mitigatespart_of, belongs_to, is_arequires_watering, survived, deterioratedExample Query 1: Find trees at compounded risk
MATCH (tree:Tree)
-[:LOCATED_NEAR]->(street:Street {type: 'motorway'})
-[:IN_DISTRICT]->(district:District)
-[:HAS_CLIMATE]->(climate:Climate {heat_days: '>30'})
WHERE tree.age > 70 AND tree.vitality = 'poor'
RETURN tree
Example Query 2: Identify intervention patterns
MATCH (tree:Tree)-[:RECEIVED_INTERVENTION]->(action:Maintenance)
-[:RESULTED_IN]->(outcome:Outcome)
WHERE outcome.vitality_improved = true
RETURN action.type, COUNT(tree) as success_count
ORDER BY success_count DESC
Temporal reasoning is the ability to understand, represent, and reason about how things change over time. It’s about answering questions like:
The term “deep” refers to using deep learning methods (neural networks) to:
Point-in-Time (Current System):
Tree 12345 at 2024-10-18:
vitality = 2
age = 75
risk = 0.83
Time Series (Historical View):
Tree 12345 over time:
2015: vitality=1, circumference=120cm, risk=0.45
2017: vitality=1, circumference=126cm, risk=0.47
2019: vitality=2, circumference=130cm, risk=0.65 ← Drought year
2021: vitality=2, circumference=131cm, risk=0.68 ← Slow growth
2024: vitality=2, circumference=133cm, risk=0.83 ← Current
Trend Detection:
Event Impact Analysis:
Trees don’t exist in isolation in time:
What they do: Process sequences by maintaining internal “memory” of past inputs
Example:
# Simplified RNN for vitality prediction
for each year in [2015, 2016, 2017, ...]:
hidden_state = RNN(
current_conditions + hidden_state_from_last_year
)
predicted_vitality = output_layer(hidden_state)
Best for: Short-term predictions, detecting gradual trends
What they do: Advanced RNNs that can remember information for longer periods
Key advantage: Can learn which past events matter most
Example use case:
"This tree's current poor health is related to the drought
from 5 years ago, not just last year's conditions"
What they do: Apply convolutional filters across time dimensions
Analogy: Like looking for patterns in a time-series “image”
Best for: Capturing multiple time scales (daily, seasonal, annual patterns)
What they do: Use attention mechanisms to focus on important time points
Key advantage: Can handle irregular sampling and long sequences
Example:
"To predict this tree's 2025 condition, pay attention to:
- 2019 drought (high importance)
- 2022 maintenance intervention (medium importance)
- Normal years (low importance)"
Combines both approaches: Represent knowledge graphs with temporal evolution
Example:
Tree_12345 ─[had_vitality(t=2019)]─> "Fair"
─[had_vitality(t=2021)]─> "Poor"
─[experienced(t=2019)]─> Drought_Event
Methods:
| Advantage | Description | Impact on BlueGreenCity-KI |
|---|---|---|
| Rich Relationship Modeling | Explicitly captures connections between trees, locations, events, and conditions | Better understanding of compound risk factors (e.g., old tree + motorway + heatwave) |
| Explainable AI | Paths through the graph show reasoning chains | Transparent decision support for urban planners |
| Domain Knowledge Integration | Can encode expert rules and scientific knowledge | Incorporate arborist expertise directly |
| Flexible Querying | Complex questions answered through graph traversal | “Find all vulnerable tree groups near critical infrastructure” |
| Incremental Updates | Add new information without reprocessing entire dataset | Real-time integration of maintenance logs, weather events |
| Multi-hop Reasoning | Can infer indirect relationships | “Trees in this area tend to fail because they’re near a factory (soil contamination) that’s 500m away” |
| Entity Resolution | Link data from multiple sources | Integrate municipal data, citizen reports, sensor networks |
| Challenge | Description | Mitigation Strategy |
|---|---|---|
| Complexity | Requires learning graph databases and query languages (Cypher, SPARQL) | Start with small pilot, provide training |
| Performance | Graph queries can be slow on large graphs (>millions of nodes) | Use optimized graph databases, careful indexing |
| Data Preparation | Converting tabular data to graph format requires careful design | Develop clear ontology, automated conversion tools |
| Maintenance Overhead | Graph schema evolution is more complex than adding columns | Use versioning, migration tools |
| Learning Curve | Team needs to learn new paradigms | Incremental adoption, documentation |
| Infrastructure | Requires dedicated graph database | Use managed services (Neo4j Aura, Neptune) |
| Quality Control | Garbage in → garbage out; relationship quality matters | Validation rules, data quality checks |
| Advantage | Description | Impact on BlueGreenCity-KI |
|---|---|---|
| Predictive Power | Learn from historical patterns to forecast future states | “This tree will likely need intervention in 18 months” |
| Automatic Pattern Discovery | Neural networks find patterns humans might miss | Discover unexpected risk factors or protective factors |
| Handling Complexity | Can model non-linear interactions and long-range dependencies | Capture lag effects (stress → visible damage after 2-3 years) |
| Irregular Data | Modern methods handle missing data and irregular sampling | Work with incomplete historical records |
| Multi-scale Analysis | Capture patterns at different time scales (daily, seasonal, annual) | Understand both acute stress and chronic deterioration |
| Continuous Learning | Models improve as more data becomes available | System gets better over time |
| Anomaly Detection | Identify unusual temporal patterns | Early warning for unexpected deterioration |
| Challenge | Description | Mitigation Strategy |
|---|---|---|
| Data Requirements | Need substantial historical data for training | Start with aggregated patterns, transfer learning |
| Black Box Nature | Neural networks are harder to interpret than weighted sums | Use attention visualization, SHAP values |
| Computational Cost | Training deep models requires significant resources | Use pre-trained models, cloud GPUs |
| Overfitting Risk | Models may memorize training data rather than learn patterns | Cross-validation, regularization techniques |
| Maintenance | Models need retraining as conditions change | Automated retraining pipelines |
| Expertise Required | Need ML/deep learning skills on the team | Collaborate with ML experts, use AutoML tools |
| Validation Challenges | Hard to validate long-term predictions | Backtesting, continuous monitoring |
| Criterion | Current System | Knowledge Graphs | Deep Temporal Reasoning | Combined Approach |
|---|---|---|---|---|
| Interpretability | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐ | ⭐⭐⭐ |
| Predictive Power | ⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐ |
| Data Requirements | ⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐ |
| Complexity | ⭐ | ⭐⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐⭐⭐⭐ |
| Relationship Modeling | ⭐ | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐⭐⭐⭐ |
| Temporal Understanding | ⭐ | ⭐⭐⭐ | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐⭐ |
| Real-time Updates | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐ | ⭐⭐⭐ |
| Domain Knowledge Integration | ⭐⭐⭐ | ⭐⭐⭐⭐⭐ | ⭐⭐ | ⭐⭐⭐⭐ |
Set up infrastructure and convert existing data to graph format.
Recommended Stack:
Graph Database: Neo4j Community Edition
- Free, open-source
- Excellent documentation
- Python driver available
- QGIS plugin for visualization
Temporal ML: PyTorch + PyTorch Geometric
- Flexible, widely used
- Good support for graph neural networks
- Extensive tutorials
Data Pipeline: Python + pandas + geopandas
- Leverage existing expertise
- Easy integration with PostGIS
Alternative (Cloud-based):
Graph Database: Amazon Neptune or Neo4j Aura
- Managed service
- Better scalability
- Higher cost
Temporal ML: AWS SageMaker or Google Vertex AI
- Managed ML platform
- Built-in monitoring
Core Ontology:
# Entity Types
:Tree a owl:Class .
:Location a owl:Class .
:Species a owl:Class .
:EnvironmentalCondition a owl:Class .
:MaintenanceEvent a owl:Class .
:RiskFactor a owl:Class .
# Spatial Relations
:locatedIn a owl:ObjectProperty .
:nearStreet a owl:ObjectProperty .
:inDistrict a owl:ObjectProperty .
# Temporal Relations
:observedAt a owl:DatatypeProperty .
:validFrom a owl:DatatypeProperty .
:validUntil a owl:DatatypeProperty .
# Risk Relations
:hasRiskFactor a owl:ObjectProperty .
:causedBy a owl:ObjectProperty .
:mitigatedBy a owl:ObjectProperty .
Example Entities and Relations:
// Trees
(:Tree {
id: "12345",
species: "Tilia cordata",
planting_year: 1949,
crown_diameter: 8.5
})
// Locations
(:Street {
id: "A2_section_5",
type: "motorway",
traffic_volume: "high",
pollution_index: 0.85
})
// Environmental Conditions
(:HeatEvent {
id: "drought_2019",
start_date: "2019-06-01",
end_date: "2019-08-31",
max_temperature: 38.5,
severity: "extreme"
})
// Relations
(:Tree {id: "12345"})-[:LOCATED_NEAR {distance_m: 5}]->(:Street {id: "A2"})
(:Tree {id: "12345"})-[:EXPOSED_TO]->(:HeatEvent {id: "drought_2019"})
(:Tree {id: "12345"})-[:HAS_VITALITY {timestamp: "2019-09", value: 2}]->()
Convert PostGIS trees to graph:
from neo4j import GraphDatabase
import geopandas as gpd
from sqlalchemy import create_engine
class TreeGraphBuilder:
def __init__(self, neo4j_uri, neo4j_user, neo4j_password):
self.driver = GraphDatabase.driver(
neo4j_uri,
auth=(neo4j_user, neo4j_password)
)
def load_trees_from_postgis(self):
"""Load trees from PostGIS"""
engine = create_engine(get_postgis_connection_string())
query = "SELECT * FROM bgcki.trees_inventory"
trees = gpd.read_postgis(query, engine, geom_col='geometry')
return trees
def create_tree_nodes(self, trees_gdf):
"""Create tree nodes in Neo4j"""
with self.driver.session() as session:
for idx, tree in trees_gdf.iterrows():
session.run("""
MERGE (t:Tree {id: $id})
SET t.species = $species,
t.planting_year = $planting_year,
t.vitality = $vitality,
t.latitude = $lat,
t.longitude = $lon,
t.geometry = point({x: $lon, y: $lat})
""",
id=tree['gisid'],
species=tree.get('art'),
planting_year=tree.get('pflanzjahr'),
vitality=tree.get('vitalitaet'),
lat=tree.geometry.y,
lon=tree.geometry.x
)
def create_spatial_relations(self, trees_gdf, streets_gdf):
"""Create spatial relationships"""
with self.driver.session() as session:
# Find trees near streets (within 10m)
for idx, tree in trees_gdf.iterrows():
nearby_streets = streets_gdf[
streets_gdf.distance(tree.geometry) < 10
]
for _, street in nearby_streets.iterrows():
session.run("""
MATCH (t:Tree {id: $tree_id})
MATCH (s:Street {id: $street_id})
MERGE (t)-[:LOCATED_NEAR {
distance_m: $distance
}]->(s)
""",
tree_id=tree['gisid'],
street_id=street['id'],
distance=tree.geometry.distance(street.geometry)
)
# Usage
builder = TreeGraphBuilder(
neo4j_uri="bolt://localhost:7687",
neo4j_user="neo4j",
neo4j_password="password"
)
trees = builder.load_trees_from_postgis()
builder.create_tree_nodes(trees)
Integrate knowledge graph queries into the existing pipeline.
Example: Enhanced Planting Structure with KG
class PlantingStructureKG(Criterion):
"""
Planting structure criterion enhanced with knowledge graph.
Considers not just spatial proximity but also:
- Shared species (same species provide better support)
- Historical co-survival patterns
- Shared maintenance history
"""
def compute(self, ctx: Context) -> CriterionResult:
trees = ctx.vector["trees"]
scores = []
# Query knowledge graph for each tree
with ctx.graph_db.session() as session:
for tree_id in trees[ctx.schema.tree_id]:
# Find support network
result = session.run("""
MATCH (t:Tree {id: $tree_id})
OPTIONAL MATCH (t)-[:NEAR]-(neighbor:Tree)
WHERE neighbor.species = t.species
WITH t, COUNT(neighbor) as same_species_neighbors
OPTIONAL MATCH (t)-[:NEAR]-(any_neighbor:Tree)
WITH t, same_species_neighbors, COUNT(any_neighbor) as total_neighbors
RETURN same_species_neighbors, total_neighbors
""", tree_id=tree_id).single()
# Compute enhanced score
same_species = result['same_species_neighbors']
total = result['total_neighbors']
# Lower risk if many same-species neighbors
if same_species >= 5:
score = 0.0 # Strong support group
elif same_species >= 2:
score = 0.3 # Moderate support
elif total >= 5:
score = 0.5 # Mixed group
else:
score = 1.0 # Isolated
scores.append(score)
return CriterionResult(
name=self.name,
score=pd.Series(scores, index=trees[ctx.schema.tree_id])
)
Use graph queries to find compound risk scenarios:
// Query: Find trees with multiple compounding risk factors
MATCH (t:Tree)
WHERE t.age > 70 // Old
AND t.vitality >= 2 // Fair or poor condition
// Near high-pollution street
MATCH (t)-[:LOCATED_NEAR {distance_m: <10}]->(s:Street)
WHERE s.type IN ['motorway', 'trunk', 'primary']
// In heat-exposed area
MATCH (t)-[:IN_MICROCLIMATE]->(m:Microclimate)
WHERE m.avg_temp_14h > 30
// Isolated planting
MATCH (t)
WHERE NOT (t)-[:NEAR]-(other:Tree)
RETURN t.id, t.species, t.age, t.vitality
ORDER BY t.age DESC
Integration into pipeline:
def detect_compound_risks(ctx: Context) -> pd.DataFrame:
"""Identify trees with multiple interacting risk factors"""
with ctx.graph_db.session() as session:
result = session.run("""
MATCH (t:Tree)
WHERE t.age > 70 AND t.vitality >= 2
MATCH (t)-[:LOCATED_NEAR {distance_m: <10}]->(s:Street)
WHERE s.type IN ['motorway', 'trunk']
MATCH (t)-[:IN_MICROCLIMATE]->(m:Microclimate)
WHERE m.avg_temp_14h > 30
OPTIONAL MATCH (t)-[:NEAR]-(neighbor:Tree)
WITH t, COUNT(neighbor) as neighbor_count
WHERE neighbor_count < 2
RETURN t.id as tree_id,
'compound_high_risk' as risk_category,
1.0 as risk_multiplier
""")
# Convert to DataFrame
compound_risks = pd.DataFrame([dict(record) for record in result])
return compound_risks
Add temporal reasoning capabilities for prediction and trend analysis.
Create temporal node versions:
class TemporalGraphBuilder:
"""Build temporal knowledge graph from historical data"""
def add_temporal_vitality_records(self, historical_df):
"""Add time-stamped vitality observations"""
with self.driver.session() as session:
for _, record in historical_df.iterrows():
session.run("""
MATCH (t:Tree {id: $tree_id})
MERGE (t)-[:HAD_VITALITY {
timestamp: datetime($timestamp),
value: $vitality
}]->()
""",
tree_id=record['BaumID'],
timestamp=record['measurement_date'].isoformat(),
vitality=record['vitalitaet']
)
def add_environmental_events(self, events_df):
"""Add drought, storm, maintenance events"""
with self.driver.session() as session:
for _, event in events_df.iterrows():
session.run("""
CREATE (e:Event {
id: $event_id,
type: $event_type,
start_date: datetime($start_date),
end_date: datetime($end_date),
severity: $severity
})
""",
event_id=event['id'],
event_type=event['type'],
start_date=event['start_date'].isoformat(),
end_date=event['end_date'].isoformat(),
severity=event['severity']
)
# Link affected trees
affected_trees = event['affected_tree_ids']
for tree_id in affected_trees:
session.run("""
MATCH (t:Tree {id: $tree_id})
MATCH (e:Event {id: $event_id})
MERGE (t)-[:EXPERIENCED {
impact: $impact
}]->(e)
""",
tree_id=tree_id,
event_id=event['id'],
impact=event.get('impact', 'unknown')
)
Train temporal model:
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
class TreeTimeSeriesDataset(Dataset):
"""Dataset for tree vitality time series"""
def __init__(self, temporal_df, sequence_length=5):
"""
Args:
temporal_df: DataFrame with columns [tree_id, year, vitality, features...]
sequence_length: Number of past years to use for prediction
"""
self.sequence_length = sequence_length
self.sequences = self._create_sequences(temporal_df)
def _create_sequences(self, df):
"""Create input-output sequences"""
sequences = []
for tree_id in df['tree_id'].unique():
tree_data = df[df['tree_id'] == tree_id].sort_values('year')
if len(tree_data) < self.sequence_length + 1:
continue # Not enough history
for i in range(len(tree_data) - self.sequence_length):
# Input: sequence_length past observations
X = tree_data.iloc[i:i+self.sequence_length][[
'vitality', 'age', 'temperature', 'precipitation'
]].values
# Output: next vitality value
y = tree_data.iloc[i+self.sequence_length]['vitality']
sequences.append((X, y, tree_id))
return sequences
def __len__(self):
return len(self.sequences)
def __getitem__(self, idx):
X, y, tree_id = self.sequences[idx]
return torch.FloatTensor(X), torch.FloatTensor([y]), tree_id
class VitalityLSTM(nn.Module):
"""LSTM model for predicting tree vitality"""
def __init__(self, input_dim=4, hidden_dim=64, num_layers=2):
super().__init__()
self.lstm = nn.LSTM(
input_size=input_dim,
hidden_size=hidden_dim,
num_layers=num_layers,
batch_first=True,
dropout=0.2
)
self.fc = nn.Sequential(
nn.Linear(hidden_dim, 32),
nn.ReLU(),
nn.Dropout(0.2),
nn.Linear(32, 1)
)
def forward(self, x):
# x shape: (batch, sequence_length, input_dim)
lstm_out, _ = self.lstm(x)
# Use last time step output
last_output = lstm_out[:, -1, :]
# Predict vitality
vitality_pred = self.fc(last_output)
return vitality_pred
def train_vitality_predictor():
"""Train LSTM model for vitality prediction"""
# Load historical data
historical_df = load_historical_vitality_data()
# Create dataset
dataset = TreeTimeSeriesDataset(historical_df, sequence_length=5)
train_loader = DataLoader(dataset, batch_size=32, shuffle=True)
# Initialize model
model = VitalityLSTM(input_dim=4, hidden_dim=64)
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
# Training loop
model.train()
for epoch in range(50):
total_loss = 0
for X, y, _ in train_loader:
optimizer.zero_grad()
y_pred = model(X)
loss = criterion(y_pred, y)
loss.backward()
optimizer.step()
total_loss += loss.item()
avg_loss = total_loss / len(train_loader)
print(f"Epoch {epoch+1}/50, Loss: {avg_loss:.4f}")
return model
def predict_future_vitality(model, tree_id, current_features, years_ahead=3):
"""Predict future vitality for a tree"""
model.eval()
with torch.no_grad():
# Use last 5 years of data
X = torch.FloatTensor(current_features).unsqueeze(0)
predictions = []
for year in range(years_ahead):
# Predict next year
y_pred = model(X)
predictions.append(y_pred.item())
# Update sequence (rolling window)
# ... (implementation details)
return predictions
For more sophisticated temporal reasoning:
import torch_geometric as pyg
from torch_geometric.nn import GATConv
class TemporalGraphAttentionNetwork(nn.Module):
"""
Temporal graph neural network with attention.
Learns which past events and relationships matter most.
"""
def __init__(self, node_features=10, hidden_dim=64, num_heads=4):
super().__init__()
# Spatial graph attention layers
self.gat1 = GATConv(node_features, hidden_dim, heads=num_heads)
self.gat2 = GATConv(hidden_dim * num_heads, hidden_dim, heads=1)
# Temporal attention
self.temporal_attention = nn.MultiheadAttention(
embed_dim=hidden_dim,
num_heads=4,
batch_first=True
)
# Prediction head
self.predictor = nn.Sequential(
nn.Linear(hidden_dim, 32),
nn.ReLU(),
nn.Linear(32, 1),
nn.Sigmoid() # Output: risk score 0-1
)
def forward(self, x, edge_index, temporal_x):
"""
Args:
x: Current node features (num_nodes, node_features)
edge_index: Graph connectivity (2, num_edges)
temporal_x: Historical features (num_nodes, seq_len, node_features)
"""
# Spatial graph processing
x = self.gat1(x, edge_index)
x = torch.relu(x)
x = self.gat2(x, edge_index)
# Temporal attention over historical states
temporal_out, attention_weights = self.temporal_attention(
query=x.unsqueeze(1), # Current state
key=temporal_x, # Historical states
value=temporal_x
)
# Combine spatial and temporal information
combined = x + temporal_out.squeeze(1)
# Predict risk
risk_score = self.predictor(combined)
return risk_score, attention_weights
Combine static scoring with graph/temporal models:
class HybridRiskAssessment:
"""
Combines traditional weighted scoring with knowledge graph
and temporal reasoning.
"""
def __init__(self, ctx: Context, kg_session, temporal_model):
self.ctx = ctx
self.kg_session = kg_session
self.temporal_model = temporal_model
def compute_risk_scores(self) -> pd.DataFrame:
"""Compute risk using hybrid approach"""
# 1. Traditional static scoring (fast, interpretable)
static_scores = self._compute_static_scores()
# 2. Knowledge graph reasoning (relationship-based adjustments)
kg_adjustments = self._compute_kg_adjustments()
# 3. Temporal predictions (future risk projection)
temporal_predictions = self._compute_temporal_predictions()
# 4. Combine all signals
final_scores = self._combine_scores(
static_scores,
kg_adjustments,
temporal_predictions
)
return final_scores
def _compute_static_scores(self) -> pd.DataFrame:
"""Traditional weighted aggregation"""
from bluegreencity_ki.pipelines import risk_scoring
results, _, _, _ = risk_scoring.run(self.ctx)
return results
def _compute_kg_adjustments(self) -> pd.Series:
"""Graph-based risk multipliers"""
adjustments = {}
trees = self.ctx.vector["trees"]
for tree_id in trees[self.ctx.schema.tree_id]:
# Query for compound risk factors
result = self.kg_session.run("""
MATCH (t:Tree {id: $tree_id})
// Check for protective factors
OPTIONAL MATCH (t)-[:NEAR]-(group:Tree)
WITH t, COUNT(group) as group_size
// Check for stress factors
OPTIONAL MATCH (t)-[:LOCATED_NEAR]->(s:Street {type: 'motorway'})
OPTIONAL MATCH (t)-[:EXPERIENCED]->(e:Event {type: 'drought'})
WHERE e.start_date > datetime() - duration('P2Y') // Last 2 years
WITH t, group_size,
COUNT(DISTINCT s) as near_motorway,
COUNT(DISTINCT e) as recent_droughts
// Compute adjustment multiplier
RETURN
CASE
WHEN group_size > 5 AND recent_droughts = 0 THEN 0.8
WHEN group_size < 2 AND near_motorway > 0 THEN 1.3
WHEN recent_droughts > 1 AND near_motorway > 0 THEN 1.5
ELSE 1.0
END as adjustment
""", tree_id=tree_id).single()
adjustments[tree_id] = result['adjustment'] if result else 1.0
return pd.Series(adjustments)
def _compute_temporal_predictions(self) -> pd.Series:
"""Predict future deterioration risk"""
# Load historical sequences for each tree
# Run LSTM model
# Return predicted risk increase over next 1-2 years
# (Implementation similar to Phase 3.2)
pass
def _combine_scores(self, static, kg_adj, temporal) -> pd.DataFrame:
"""Intelligently combine different risk signals"""
result = pd.DataFrame({
'tree_id': static.index,
'static_risk': static['overall_drought_risk'],
'kg_adjustment': kg_adj,
'temporal_trend': temporal,
})
# Weighted combination
result['final_risk'] = (
0.5 * result['static_risk'] * result['kg_adjustment'] +
0.3 * result['static_risk'] +
0.2 * result['temporal_trend']
).clip(0, 1)
# Add explanation
result['risk_factors'] = result.apply(
lambda row: self._explain_risk(row), axis=1
)
return result
def _explain_risk(self, row) -> str:
"""Generate human-readable risk explanation"""
explanations = []
if row['static_risk'] > 0.7:
explanations.append("High baseline risk from current conditions")
if row['kg_adjustment'] > 1.2:
explanations.append("Compound risk factors detected")
if row['temporal_trend'] > 0.6:
explanations.append("Predicted deterioration over next 2 years")
return "; ".join(explanations) if explanations else "Standard risk profile"
Set up continuous monitoring:
class HybridModelMonitor:
"""Monitor performance of hybrid risk assessment"""
def __init__(self, db_connection):
self.db = db_connection
def track_predictions(self, predictions_df):
"""Store predictions for later validation"""
predictions_df['prediction_date'] = datetime.now()
predictions_df.to_sql(
'risk_predictions_history',
self.db,
if_exists='append'
)
def validate_predictions(self):
"""Compare predictions from 1 year ago with actual outcomes"""
query = """
SELECT
p.tree_id,
p.predicted_risk,
p.prediction_date,
c.actual_vitality,
c.measurement_date,
CASE
WHEN p.predicted_risk > 0.7 AND c.actual_vitality >= 3 THEN 'true_positive'
WHEN p.predicted_risk > 0.7 AND c.actual_vitality < 3 THEN 'false_positive'
WHEN p.predicted_risk <= 0.7 AND c.actual_vitality >= 3 THEN 'false_negative'
ELSE 'true_negative'
END as prediction_outcome
FROM risk_predictions_history p
JOIN current_vitality c ON p.tree_id = c.tree_id
WHERE p.prediction_date < NOW() - INTERVAL '1 year'
AND c.measurement_date > p.prediction_date
"""
results = pd.read_sql(query, self.db)
# Compute metrics
accuracy = (results['prediction_outcome'].isin([
'true_positive', 'true_negative'
])).mean()
print(f"1-year prediction accuracy: {accuracy:.2%}")
return results
Scenario: Finding trees at indirect risk
Traditional Query (SQL):
SELECT * FROM trees
WHERE age > 70 AND vitality = 'poor'
AND street_type = 'motorway'
Knowledge Graph Query (Cypher):
// Find trees at compound risk via indirect relationships
MATCH (t:Tree)
WHERE t.age > 70 AND t.vitality >= 2
// Direct exposure to pollution
MATCH (t)-[:LOCATED_NEAR {distance_m: <10}]->(s:Street)
WHERE s.type = 'motorway'
// Also near industrial area (indirect exposure)
MATCH (s)-[:IN_DISTRICT]->(d:District)
-[:CONTAINS]->(factory:Factory)
WHERE factory.emission_class = 'high'
// In a drought-prone microclimate
MATCH (t)-[:IN_MICROCLIMATE]->(m:Microclimate)
WHERE m.historical_drought_frequency > 0.3
// With history of stress
MATCH (t)-[:EXPERIENCED]->(past_event:Event {type: 'drought'})
WHERE past_event.year >= 2018
RETURN t.id,
t.species,
'compounded_indirect_risk' as category,
factory.name as indirect_stressor
Value: Discovers trees that might be missed by simple queries because risk factors are not directly adjacent.
Scenario: Detect slow-developing vs. acute stress
def classify_stress_pattern(tree_history):
"""
Classify whether tree deterioration is:
- Acute: Sudden decline (e.g., after storm)
- Chronic: Gradual deterioration (e.g., pollution)
- Cyclical: Seasonal or periodic stress
"""
# Extract vitality time series
vitality_series = tree_history['vitality'].values
years = tree_history['year'].values
# Compute rate of change
changes = np.diff(vitality_series)
# Detect patterns
if any(changes > 1.5): # Sudden jump in vitality stage
return "acute_stress", find_trigger_event(tree_history)
elif np.all(changes >= 0) and np.mean(changes) > 0.2:
return "chronic_deterioration", "gradual_decline"
elif detect_cyclical_pattern(changes):
return "cyclical_stress", "seasonal_vulnerability"
else:
return "stable", "no_pattern"
def find_trigger_event(tree_history):
"""Find external event that triggered acute stress"""
# Query knowledge graph for events
query = """
MATCH (t:Tree {id: $tree_id})-[:EXPERIENCED]->(e:Event)
WHERE e.date >= datetime($stress_date) - duration('P3M')
AND e.date <= datetime($stress_date) + duration('P1M')
RETURN e.type, e.severity, e.date
ORDER BY e.severity DESC
LIMIT 1
"""
# Execute and return likely trigger
# ...
Scenario: Optimize inspection and intervention timing
class PredictiveMaintenanceOptimizer:
"""
Use temporal models to schedule tree inspections and interventions
at optimal times.
"""
def prioritize_inspections(self, budget_constraints):
"""
Generate inspection priority list based on:
- Current risk score
- Predicted deterioration rate
- Uncertainty in predictions (inspect trees we're unsure about)
- Maintenance cost and feasibility
"""
trees = self.load_current_tree_inventory()
priorities = []
for tree in trees:
# Current risk
current_risk = tree['overall_drought_risk']
# Predicted future risk (1 year ahead)
future_risk = self.temporal_model.predict(
tree_id=tree['id'],
horizon_years=1
)
# Deterioration rate
deterioration_rate = future_risk - current_risk
# Model uncertainty (higher uncertainty → higher priority to inspect)
prediction_uncertainty = self.temporal_model.get_uncertainty(
tree_id=tree['id']
)
# Maintenance cost from KG
maintenance_cost = self.kg_query_maintenance_cost(tree['id'])
# Priority score
priority = (
0.3 * current_risk +
0.4 * deterioration_rate +
0.2 * prediction_uncertainty +
0.1 * (1 / maintenance_cost) # Lower cost → higher priority
)
priorities.append({
'tree_id': tree['id'],
'priority_score': priority,
'current_risk': current_risk,
'predicted_1yr_risk': future_risk,
'deterioration_rate': deterioration_rate,
'recommended_action': self.recommend_action(
current_risk, future_risk
)
})
# Sort by priority
priorities_df = pd.DataFrame(priorities).sort_values(
'priority_score', ascending=False
)
# Apply budget constraint
return self.apply_budget_constraint(
priorities_df,
budget_constraints
)
def recommend_action(self, current_risk, future_risk):
"""Recommend specific intervention"""
if current_risk > 0.8:
return "immediate_watering_required"
elif future_risk > 0.8 and current_risk < 0.6:
return "monitor_closely_preventive_action"
elif future_risk - current_risk > 0.3:
return "intervention_within_6_months"
else:
return "standard_monitoring"
Scenario: Learn which species thrive under which conditions
// Query: Which species are most resilient in hot microclimates?
MATCH (t:Tree)-[:IS_SPECIES]->(s:Species)
WHERE t.age > 30 // Mature trees only
// Trees in hot areas
MATCH (t)-[:IN_MICROCLIMATE]->(m:Microclimate)
WHERE m.avg_temp_14h > 31
// Survived recent droughts
MATCH (t)-[:EXPERIENCED]->(drought:Event {type: 'drought'})
WHERE drought.year >= 2018
AND t.vitality <= 2 // Still healthy
// Group by species
WITH s.name as species_name,
COUNT(t) as survivor_count,
AVG(t.vitality) as avg_vitality
ORDER BY survivor_count DESC
RETURN species_name,
survivor_count,
avg_vitality,
'heat_resilient' as profile
LIMIT 10
Use this to:
Goal: Demonstrate value without disrupting current production system
Recommended Actions:
Parallel Operation:
Incremental Adoption:
Validation Gates:
Knowledge Graphs:
Temporal Reasoning:
Temporal Knowledge Graphs:
Urban Tree Management:
Graph Databases:
Deep Learning for Time Series:
Graph Neural Networks:
Tutorials and Courses:
Graph Databases:
Visualization:
ML Platforms:
Similar Applications:
For BlueGreenCity-KI Team:
Dynamic knowledge graphs and deep temporal reasoning offer powerful complementary approaches to the current static risk scoring pipeline. Rather than replacing the existing system, we recommend a phased, hybrid integration that:
Key Takeaway: Start small with a pilot project, validate thoroughly, and scale gradually. The goal is not perfection but progressive improvement in urban tree management outcomes.
Document Version: 1.0
Last Updated: January 2026
Authors: BlueGreenCity-KI Research Team
Status: For review and discussion