TNFR Self-Optimizing Mathematical Engine
This module implements self-optimization that emerges naturally from analyzing the mathematical structure of the nodal equation ∂EPI/∂t = νf · ΔNFR(t).
Mathematical Foundation: The nodal equation reveals natural optimization landscapes:
Self-Optimization Mechanisms:
Status: CANONICAL SELF-OPTIMIZING ENGINE
"""
TNFR Self-Optimizing Mathematical Engine
This module implements self-optimization that emerges naturally from analyzing
the mathematical structure of the nodal equation ∂EPI/∂t = νf · ΔNFR(t).
Mathematical Foundation:
The nodal equation reveals natural optimization landscapes:
1. **Gradient Flows**: ΔNFR naturally defines optimization directions
2. **Energy Functionals**: EPI configurations have natural energy measures
3. **Constraint Manifolds**: Grammar rules create constraint manifolds
4. **Variational Principles**: Operator sequences minimize action functionals
5. **Learning Dynamics**: Repeated patterns improve through experience
6. **Adaptive Algorithms**: The system learns optimal strategies automatically
Self-Optimization Mechanisms:
- Automatic cache strategy learning based on mathematical importance
- Dynamic operator sequence optimization using variational principles
- Adaptive precision management based on mathematical requirements
- Self-tuning computational backend selection
- Emergent load balancing through mathematical analysis
- Natural parallelization discovery via spectral decomposition
Status: CANONICAL SELF-OPTIMIZING ENGINE
"""
import hashlib
import json
import re
import threading
import time
from collections import defaultdict, deque
from dataclasses import dataclass
from datetime import datetime, timezone
from enum import Enum
from pathlib import Path
from statistics import fmean
from typing import Any, Mapping, Sequence
from ..alias import get_attr
from ..constants.aliases import ALIAS_DNFR, ALIAS_VF
# Operational engine-tuning knobs (not TNFR physics) → tnfr.constants.operational
from ..constants.operational import (
NODAL_OPT_COUPLING_CANONICAL,
)
from ..constants.canonical import PI # π ≈ 3.1416 (structural scale)
from ..constants.operational import (
SELF_OPT_CACHE_CONTRACTION_CANONICAL,
SELF_OPT_CACHE_EXPANSION_CANONICAL,
SELF_OPT_CACHE_HIGH_FRACTION_CANONICAL,
SELF_OPT_CACHE_LOW_FRACTION_CANONICAL,
SELF_OPT_CHIRALITY_THRESHOLD_CANONICAL,
SELF_OPT_COMPRESSION_HIGH_CANONICAL,
SELF_OPT_COUPLING_LOW_CANONICAL,
SELF_OPT_DENSITY_DENSE_CANONICAL,
SELF_OPT_DENSITY_SPARSE_CANONICAL,
SELF_OPT_ENERGY_HIGH_CANONICAL,
SELF_OPT_IMPROVEMENT_SIGNIFICANT_CANONICAL,
SELF_OPT_SPEEDUP_HIGH_CANONICAL,
SELF_OPT_SPEEDUP_LOW_CANONICAL,
SELF_OPT_SYMMETRY_THRESHOLD_CANONICAL,
)
from ..errors import TNFRValueError
from ..mathematics.unified_numerical import np
from ..operators.grammar import glyph_function_name, validate_sequence
try:
import networkx as nx
HAS_NETWORKX = True
except ImportError:
HAS_NETWORKX = False
nx = None
HAS_SCIPY = True # Assume available for mathematical analysis
def _extract_scalar_epi(val: Any) -> float:
"""Extract scalar magnitude from potentially complex/dict EPI value."""
if isinstance(val, (int, float)):
return float(val)
if isinstance(val, complex):
return float(np.abs(val))
if isinstance(val, dict):
if "continuous" in val:
c = val["continuous"]
if isinstance(c, (tuple, list)) and len(c) > 0:
v = c[0]
return float(np.abs(v)) if isinstance(v, complex) else float(v)
return 0.0
# Import Unified Fields (New Nov 2025)
try:
from ..physics.fields import compute_unified_telemetry
HAS_UNIFIED_FIELDS = True
except ImportError:
HAS_UNIFIED_FIELDS = False
# Import Structural Integrity Monitor (closed-loop conservation)
try:
from ..physics.integrity import StructuralIntegrityMonitor
HAS_INTEGRITY_MONITOR = True
except ImportError:
HAS_INTEGRITY_MONITOR = False
# Import conservation functions for closed-loop optimization (P5)
try:
from ..physics.conservation import (
capture_conservation_snapshot,
compute_lyapunov_derivative,
detect_grammar_violations_from_conservation,
verify_conservation_balance,
)
HAS_CONSERVATION = True
except ImportError:
HAS_CONSERVATION = False
try:
from ..metrics.common import compute_coherence
from ..metrics.sense_index import compute_Si
HAS_METRIC_OPERATORS = True
except ImportError: # pragma: no cover - optional dependency in trimmed builds
HAS_METRIC_OPERATORS = False
compute_coherence = None # type: ignore
compute_Si = None # type: ignore
# Import TNFR engines
try:
from ..engines.pattern_discovery.mathematical_patterns import (
EmergentPatternType,
TNFREmergentPatternEngine,
)
from .optimization_orchestrator import (
OptimizationStrategy,
TNFROptimizationOrchestrator,
)
from .unified_backend import TNFRUnifiedBackend
HAS_ENGINES = True
except ImportError:
HAS_ENGINES = False
HAS_MATH_BACKENDS = True # Assume available
_SAFE_LABEL_RE = re.compile(r"[^A-Za-z0-9._-]+")
_DEFAULT_OUTPUT_DIR = Path("results") / "self_optimization"
# --- Self-optimization safety thresholds ---
_MIN_CONSERVATION_QUALITY = 0.7 # below → add stabilizers
_MAX_CHARGE_DRIFT = 0.1 # above → Noether charge drift correction
_MAX_VIOLATION_RATE = 0.2 # above → grammar review needed
_MIN_EPI_VARIANCE = 0.01 # below → variance too low, optimize
def _sanitize_label(value: Any | None, default: str) -> str:
"""Convert arbitrary identifiers to filesystem-safe labels."""
if value is None:
text = default
else:
text = str(value).strip()
if not text:
text = default
sanitized = _SAFE_LABEL_RE.sub("_", text)
return sanitized[:64] or default
def _mean_numeric(values: Sequence[float]) -> float | None:
"""Compute mean for numeric sequences with graceful fallback."""
data = [float(v) for v in values if isinstance(v, (int, float))]
if not data:
return None
try:
return fmean(data)
except Exception:
return float(sum(data) / len(data))
def _sense_index_mean(payload: Any) -> float | None:
"""Reduce compute_Si outputs (dict, array) to a scalar average."""
if payload is None:
return None
if isinstance(payload, dict):
return _mean_numeric(list(payload.values()))
try:
array = np.asarray(payload, dtype=float)
except Exception:
return None
if array.size == 0:
return None
return float(np.mean(array))
def _json_safe(value: Any) -> Any:
"""Convert complex objects (NumPy, mappings) to JSON-safe structures."""
if isinstance(value, (str, int, float, bool)) or value is None:
return value
if isinstance(value, np.generic):
return value.item()
if isinstance(value, np.ndarray):
return value.tolist()
if isinstance(value, Mapping):
return {str(k): _json_safe(v) for k, v in value.items()}
if isinstance(value, (list, tuple, set, frozenset)):
return [_json_safe(v) for v in value]
if hasattr(value, "__dict__"):
return {str(k): _json_safe(v) for k, v in vars(value).items()}
return repr(value)
def _canonicalize_sequence_tokens(sequence: Sequence[Any]) -> list[str]:
"""Normalize operator or glyph tokens into canonical operator names."""
tokens: list[str] = []
for entry in sequence:
if entry is None:
continue
candidate = getattr(entry, "name", entry)
text = str(candidate).strip()
if not text:
continue
canonical = glyph_function_name(text, default=None)
normalized = canonical or text
tokens.append(normalized.lower())
return tokens
class OptimizationObjective(Enum):
"""Self-optimization objectives."""
MINIMIZE_COMPUTATION_TIME = "minimize_time"
MAXIMIZE_CACHE_EFFICIENCY = "maximize_cache"
MINIMIZE_MEMORY_USAGE = "minimize_memory"
MAXIMIZE_ACCURACY = "maximize_accuracy"
MINIMIZE_ENERGY = "minimize_energy"
MAXIMIZE_THROUGHPUT = "maximize_throughput"
BALANCE_ALL = "balance_all"
class LearningStrategy(Enum):
"""Learning strategies for optimization."""
GRADIENT_DESCENT = "gradient_descent"
EVOLUTIONARY = "evolutionary"
REINFORCEMENT = "reinforcement"
BAYESIAN_OPTIMIZATION = "bayesian"
MATHEMATICAL_ANALYSIS = "mathematical"
HYBRID = "hybrid"
@dataclass
class OptimizationExperience:
"""Experience record for learning."""
graph_properties: dict[str, Any]
operation_type: str
strategy_used: str
parameters: dict[str, Any]
performance_metrics: dict[str, float]
timestamp: float
success: bool
mathematical_signature: dict[str, Any] | None = None
@dataclass
class OptimizationPolicy:
"""Learned optimization policy."""
policy_name: str
objective: OptimizationObjective
conditions: dict[str, Any] # When to apply this policy
actions: dict[str, Any] # What to do
confidence: float
success_rate: float
average_improvement: float
applications_count: int = 0
@dataclass
class SelfOptimizationResult:
"""Result of self-optimization analysis."""
learned_policies: list[OptimizationPolicy]
optimization_improvements: dict[str, float]
recommended_strategies: list[str]
mathematical_insights: dict[str, Any]
predicted_speedups: dict[str, float]
adaptive_configurations: dict[str, Any]
execution_time: float
conservation_feedback: dict[str, float] | None = None
class TNFRSelfOptimizingEngine:
"""
Self-optimizing engine that learns optimal strategies from mathematical structure.
This engine discovers optimization patterns by analyzing the mathematical
properties of the nodal equation and learning from experience.
"""
def __init__(
self,
learning_strategy: LearningStrategy = LearningStrategy.MATHEMATICAL_ANALYSIS,
optimization_objective: OptimizationObjective = OptimizationObjective.BALANCE_ALL,
max_experience_history: int = 1000,
):
self.learning_strategy = learning_strategy
self.optimization_objective = optimization_objective
self.max_experience_history = max_experience_history
# Learning state
self.experience_history: deque = deque(maxlen=max_experience_history)
self.learned_policies: list[OptimizationPolicy] = []
self.mathematical_insights: dict[str, Any] = {}
# Performance tracking
self.optimization_attempts = 0
self.successful_optimizations = 0
self.cumulative_improvements = defaultdict(float)
# Adaptive configuration
self.adaptive_config = {
"cache_size_mb": 256.0,
"precision": "float64",
"backend_preference": "numpy",
"parallel_threshold": 50,
"spectral_threshold": 20,
}
# Thread safety
self._lock = threading.Lock()
# Initialize engines if available
if HAS_ENGINES:
self.unified_backend = TNFRUnifiedBackend()
self.orchestrator = TNFROptimizationOrchestrator()
self.pattern_engine = TNFREmergentPatternEngine()
else:
self.unified_backend = None
self.orchestrator = None
self.pattern_engine = None
def analyze_mathematical_optimization_landscape(
self, G: Any, operation_type: str = "general"
) -> dict[str, Any]:
"""
Analyze mathematical structure to discover optimization opportunities.
Uses the mathematical properties of the nodal equation to identify
natural optimization strategies.
"""
insights = {}
if not HAS_NETWORKX or G is None:
return insights
# Basic graph properties
num_nodes = len(G.nodes())
num_edges = len(G.edges())
density = (
(2 * num_edges) / (num_nodes * (num_nodes - 1)) if num_nodes > 1 else 0
)
# Unified Field Analysis (New Nov 2025)
unified_insights = {}
if HAS_UNIFIED_FIELDS:
try:
# Compute full unified telemetry
telemetry = compute_unified_telemetry(G)
# Extract Complex Field
complex_data = telemetry.get("complex_field", {})
psi_mag_array = complex_data.get("magnitude", np.array([]))
psi_scalar = np.mean(psi_mag_array) if len(psi_mag_array) > 0 else 0.0
# Extract Emergent Fields
emergent_data = telemetry.get("emergent_fields", {})
chi_array = emergent_data.get("chirality", np.array([]))
sb_array = emergent_data.get("symmetry_breaking", np.array([]))
cc_array = emergent_data.get("coherence_coupling", np.array([]))
chi_scalar = np.mean(np.abs(chi_array)) if len(chi_array) > 0 else 0.0
sb_scalar = np.mean(sb_array) if len(sb_array) > 0 else 0.0
cc_scalar = np.mean(cc_array) if len(cc_array) > 0 else 0.0
# Extract Tensor Invariants
tensor_data = telemetry.get("tensor_invariants", {})
ed_array = tensor_data.get("energy_density", np.array([]))
tc_array = tensor_data.get("topological_charge", np.array([]))
ed_scalar = np.mean(ed_array) if len(ed_array) > 0 else 0.0
tc_scalar = np.mean(np.abs(tc_array)) if len(tc_array) > 0 else 0.0
unified_insights = {
"psi_magnitude": float(psi_scalar),
"chirality": float(chi_scalar),
"symmetry_breaking": float(sb_scalar),
"coherence_coupling": float(cc_scalar),
"energy_density": float(ed_scalar),
"topological_charge": float(tc_scalar),
}
insights["unified_field_analysis"] = unified_insights
except Exception as e:
# Fallback if computation fails
insights["unified_field_error"] = str(e)
# Conservation Integrity Feedback (closed-loop)
if HAS_INTEGRITY_MONITOR:
monitor = StructuralIntegrityMonitor.get(G) if G is not None else None
if monitor is not None:
fv = monitor.feedback_vector()
insights["conservation_feedback"] = fv
# Mathematical structure analysis
insights["graph_structure"] = {
"nodes": num_nodes,
"edges": num_edges,
"density": density,
"is_connected": nx.is_connected(G) if HAS_NETWORKX else False,
"avg_degree": 2 * num_edges / num_nodes if num_nodes > 0 else 0,
}
# Spectral properties for optimization
if self.pattern_engine:
pattern_result = self.pattern_engine.discover_all_patterns(G)
# Extract optimization hints from discovered patterns
optimization_hints = []
for pattern in pattern_result.discovered_patterns:
if pattern.compression_ratio > SELF_OPT_COMPRESSION_HIGH_CANONICAL:
optimization_hints.append(
f"use_compression_{pattern.pattern_type.value}"
)
if pattern.prediction_horizon > PI:
optimization_hints.append(
f"use_prediction_{pattern.pattern_type.value}"
)
if pattern.pattern_type == EmergentPatternType.EIGENMODE_RESONANCE:
optimization_hints.append("use_spectral_methods")
if pattern.pattern_type == EmergentPatternType.FRACTAL_SCALING:
optimization_hints.append("use_hierarchical_methods")
insights["pattern_optimization_hints"] = optimization_hints
insights["mathematical_patterns"] = len(pattern_result.discovered_patterns)
insights["compression_potential"] = pattern_result.compression_potential
# Nodal equation analysis
epi_values = [
_extract_scalar_epi(G.nodes[node].get("EPI", 0.0)) for node in G.nodes()
]
vf_values = [get_attr(G.nodes[node], ALIAS_VF, 1.0) for node in G.nodes()]
dnfr_values = [get_attr(G.nodes[node], ALIAS_DNFR, 0.0) for node in G.nodes()]
# Mathematical properties for optimization
epi_variance = np.var(epi_values)
vf_range = np.max(vf_values) - np.min(vf_values) if vf_values else 0
dnfr_magnitude = np.mean(np.abs(dnfr_values)) if dnfr_values else 0
# Optimization recommendations based on mathematical properties
recommendations = []
# Unified Field Recommendations (New Nov 2025)
if "unified_field_analysis" in insights:
ufa = insights["unified_field_analysis"]
# Chirality-based optimization
if abs(ufa.get("chirality", 0)) > SELF_OPT_CHIRALITY_THRESHOLD_CANONICAL:
recommendations.append("use_chiral_optimization")
# Symmetry breaking handling
if ufa.get("symmetry_breaking", 0) > SELF_OPT_SYMMETRY_THRESHOLD_CANONICAL:
recommendations.append("use_phase_transition_handling")
# Coherence coupling optimization
if ufa.get("coherence_coupling", 0) < SELF_OPT_COUPLING_LOW_CANONICAL:
recommendations.append("enhance_coherence_coupling")
# Topological charge handling
if abs(ufa.get("topological_charge", 0)) > NODAL_OPT_COUPLING_CANONICAL:
recommendations.append("topological_defect_correction")
# Energy density optimization
if ufa.get("energy_density", 0) > SELF_OPT_ENERGY_HIGH_CANONICAL:
recommendations.append("high_energy_stabilization")
# Conservation-based recommendations (closed-loop)
cf = insights.get("conservation_feedback")
if cf is not None:
if cf.get("conservation_quality", 1.0) < _MIN_CONSERVATION_QUALITY:
recommendations.append("conservation_quality_low_stabilize")
if cf.get("energy_derivative", 0.0) > 0:
recommendations.append("lyapunov_unstable_add_IL")
if cf.get("charge_drift", 0.0) > _MAX_CHARGE_DRIFT:
recommendations.append("noether_charge_drift_correction")
if cf.get("violation_rate", 0.0) > _MAX_VIOLATION_RATE:
recommendations.append("high_violation_rate_grammar_review")
if epi_variance < _MIN_EPI_VARIANCE:
recommendations.append("low_variance_epi_optimization")
if vf_range < NODAL_OPT_COUPLING_CANONICAL:
recommendations.append("uniform_vf_optimization")
if dnfr_magnitude > 1.0:
recommendations.append("high_dnfr_stabilization")
if num_nodes > 100:
recommendations.append("large_graph_optimization")
if density > SELF_OPT_DENSITY_DENSE_CANONICAL:
recommendations.append("dense_graph_optimization")
elif density < SELF_OPT_DENSITY_SPARSE_CANONICAL:
recommendations.append("sparse_graph_optimization")
insights["nodal_equation_analysis"] = {
"epi_variance": epi_variance,
"vf_range": vf_range,
"dnfr_magnitude": dnfr_magnitude,
"optimization_recommendations": recommendations,
}
return insights
def learn_from_experience(self, experience: OptimizationExperience) -> None:
"""
Learn optimization strategies from performance experience.
Uses mathematical analysis to extract general principles.
"""
with self._lock:
self.experience_history.append(experience)
self.optimization_attempts += 1
if experience.success:
self.successful_optimizations += 1
# Analyze experience for patterns
if len(self.experience_history) >= 10: # Minimum data for learning
self._extract_optimization_policies()
self._update_adaptive_configuration()
def _extract_optimization_policies(self) -> None:
"""Extract general optimization policies from experience."""
# Group experiences by similar conditions
condition_groups = defaultdict(list)
for exp in self.experience_history:
if exp.success:
# Create condition signature
graph_size = exp.graph_properties.get("nodes", 0)
density = exp.graph_properties.get("density", 0.0)
operation = exp.operation_type
# Discretize conditions for pattern recognition
size_bucket = (
"small"
if graph_size < 20
else "medium" if graph_size < 100 else "large"
)
density_bucket = (
"sparse"
if density < SELF_OPT_DENSITY_SPARSE_CANONICAL
else (
"medium"
if density < SELF_OPT_DENSITY_DENSE_CANONICAL
else "dense"
)
)
condition_key = (size_bucket, density_bucket, operation)
condition_groups[condition_key].append(exp)
# Extract policies from groups with sufficient data
new_policies = []
for condition_key, experiences in condition_groups.items():
if len(experiences) >= 3: # Minimum for reliable pattern
size_bucket, density_bucket, operation = condition_key
# Find best strategy for this condition
strategy_performance = defaultdict(list)
for exp in experiences:
strategy = exp.strategy_used
improvement = exp.performance_metrics.get("speedup_factor", 1.0)
strategy_performance[strategy].append(improvement)
# Select best strategy
best_strategy = None
best_avg_improvement = 0
for strategy, improvements in strategy_performance.items():
avg_improvement = np.mean(improvements)
if avg_improvement > best_avg_improvement:
best_avg_improvement = avg_improvement
best_strategy = strategy
if (
best_strategy
and best_avg_improvement
> SELF_OPT_IMPROVEMENT_SIGNIFICANT_CANONICAL
): # Significant improvement
policy = OptimizationPolicy(
policy_name=f"{size_bucket}_{density_bucket}_{operation}_policy",
objective=self.optimization_objective,
conditions={
"graph_size_bucket": size_bucket,
"density_bucket": density_bucket,
"operation_type": operation,
},
actions={
"recommended_strategy": best_strategy,
"expected_improvement": best_avg_improvement,
},
confidence=min(len(experiences) / 10.0, 1.0),
success_rate=len(experiences)
/ max(
1,
len(
[
e
for e in self.experience_history
if self._matches_conditions(e, condition_key)
]
),
),
average_improvement=best_avg_improvement,
)
new_policies.append(policy)
# Update learned policies
self.learned_policies.extend(new_policies)
# Remove outdated policies (keep only best 20)
if len(self.learned_policies) > 20:
self.learned_policies.sort(
key=lambda p: p.confidence * p.average_improvement, reverse=True
)
self.learned_policies = self.learned_policies[:20]
def _matches_conditions(
self, experience: OptimizationExperience, condition_key: tuple
) -> bool:
"""Check if experience matches condition key."""
size_bucket, density_bucket, operation = condition_key
graph_size = experience.graph_properties.get("nodes", 0)
density = experience.graph_properties.get("density", 0.0)
exp_size_bucket = (
"small" if graph_size < 20 else "medium" if graph_size < 100 else "large"
)
exp_density_bucket = (
"sparse"
if density < SELF_OPT_DENSITY_SPARSE_CANONICAL
else "medium" if density < SELF_OPT_DENSITY_DENSE_CANONICAL else "dense"
)
return (
exp_size_bucket == size_bucket
and exp_density_bucket == density_bucket
and experience.operation_type == operation
)
def _update_adaptive_configuration(self) -> None:
"""Update adaptive configuration based on learned patterns."""
if not self.experience_history:
return
# Analyze recent performance
recent_experiences = list(self.experience_history)[-50:] # Last 50 experiences
successful_experiences = [e for e in recent_experiences if e.success]
if not successful_experiences:
return
# Update cache size based on memory vs performance tradeoff
memory_usage = [
e.performance_metrics.get("memory_used_mb", 0)
for e in successful_experiences
]
speedups = [
e.performance_metrics.get("speedup_factor", 1.0)
for e in successful_experiences
]
if len(memory_usage) > 0 and len(speedups) > 0:
# Simple heuristic: increase cache if low memory usage but good speedup
avg_memory = np.mean(memory_usage)
avg_speedup = np.mean(speedups)
if (
avg_memory
< self.adaptive_config["cache_size_mb"]
* SELF_OPT_CACHE_LOW_FRACTION_CANONICAL
and avg_speedup > SELF_OPT_SPEEDUP_HIGH_CANONICAL
):
self.adaptive_config[
"cache_size_mb"
] *= SELF_OPT_CACHE_EXPANSION_CANONICAL
elif (
avg_memory
> self.adaptive_config["cache_size_mb"]
* SELF_OPT_CACHE_HIGH_FRACTION_CANONICAL
and avg_speedup < SELF_OPT_SPEEDUP_LOW_CANONICAL
):
self.adaptive_config[
"cache_size_mb"
] *= SELF_OPT_CACHE_CONTRACTION_CANONICAL
# Update backend preference
backend_performance = defaultdict(list)
for exp in successful_experiences:
backend = exp.parameters.get("backend", "numpy")
speedup = exp.performance_metrics.get("speedup_factor", 1.0)
backend_performance[backend].append(speedup)
if backend_performance:
best_backend = max(
backend_performance.keys(),
key=lambda b: np.mean(backend_performance[b]),
)
self.adaptive_config["backend_preference"] = best_backend
# Track conservation health across experiences (P5)
conservation_drifts = [
e.performance_metrics["conservation_charge_drift"]
for e in recent_experiences
if "conservation_charge_drift" in e.performance_metrics
]
if conservation_drifts:
self.adaptive_config["mean_conservation_drift"] = float(
np.mean(conservation_drifts)
)
def recommend_optimization_strategy(
self,
G: Any,
operation_type: str = "general",
current_performance: dict[str, float] | None = None,
) -> SelfOptimizationResult:
"""
Recommend optimization strategy based on mathematical analysis and learning.
"""
start_time = time.perf_counter()
# Analyze mathematical structure
mathematical_insights = self.analyze_mathematical_optimization_landscape(
G, operation_type
)
# Find matching learned policies
matching_policies = []
if HAS_NETWORKX and G:
num_nodes = len(G.nodes())
num_edges = len(G.edges())
density = (
(2 * num_edges) / (num_nodes * (num_nodes - 1)) if num_nodes > 1 else 0
)
size_bucket = (
"small" if num_nodes < 20 else "medium" if num_nodes < 100 else "large"
)
density_bucket = (
"sparse"
if density < SELF_OPT_DENSITY_SPARSE_CANONICAL
else "medium" if density < SELF_OPT_DENSITY_DENSE_CANONICAL else "dense"
)
for policy in self.learned_policies:
conditions = policy.conditions
if (
conditions.get("graph_size_bucket") == size_bucket
and conditions.get("density_bucket") == density_bucket
and conditions.get("operation_type") == operation_type
):
matching_policies.append(policy)
# Generate recommendations
recommended_strategies = []
predicted_speedups = {}
# From learned policies
for policy in matching_policies:
strategy = policy.actions.get("recommended_strategy")
if strategy:
recommended_strategies.append(strategy)
predicted_speedups[strategy] = policy.average_improvement
# From mathematical analysis
math_recommendations = mathematical_insights.get(
"nodal_equation_analysis", {}
).get("optimization_recommendations", [])
recommended_strategies.extend(math_recommendations)
# From pattern analysis
pattern_hints = mathematical_insights.get("pattern_optimization_hints", [])
recommended_strategies.extend(pattern_hints)
# Remove duplicates while preserving order
recommended_strategies = list(dict.fromkeys(recommended_strategies))
# Conservation-aware strategy reordering (P5: closed-loop)
# When conservation is stressed, prefer safe computational strategies
# to avoid aggressive optimizations that may degrade structural integrity.
cf = mathematical_insights.get("conservation_feedback")
if cf is not None:
cq = cf.get("conservation_quality", 1.0)
de_dt = cf.get("energy_derivative", 0.0)
if cq < _MIN_CONSERVATION_QUALITY or de_dt > 0:
safe = []
other = []
for s in recommended_strategies:
if any(
kw in s.lower()
for kw in ("cache", "structural", "stabiliz", "memo")
):
safe.append(s)
else:
other.append(s)
recommended_strategies = safe + other
# Calculate optimization improvements
optimization_improvements = {}
if current_performance:
for strategy, speedup in predicted_speedups.items():
optimization_improvements[strategy] = (speedup - 1.0) / speedup
execution_time = time.perf_counter() - start_time
return SelfOptimizationResult(
learned_policies=matching_policies,
optimization_improvements=optimization_improvements,
recommended_strategies=recommended_strategies,
mathematical_insights=mathematical_insights,
predicted_speedups=predicted_speedups,
adaptive_configurations=dict(self.adaptive_config),
execution_time=execution_time,
conservation_feedback=cf,
)
def optimize_automatically(
self, G: Any, operation_type: str = "general", **kwargs
) -> dict[str, Any]:
"""
Automatically apply best optimization strategy.
Uses learned policies and mathematical analysis to select and apply
the optimal strategy.
"""
exec_kwargs = dict(kwargs)
dry_run = bool(exec_kwargs.pop("dry_run", False))
capture_snapshots = bool(exec_kwargs.pop("capture_snapshots", dry_run))
seed_value = exec_kwargs.pop("seed", exec_kwargs.pop("random_seed", None))
node_label = (
exec_kwargs.pop("node", None)
or exec_kwargs.pop("node_id", None)
or exec_kwargs.pop("target_node", None)
or exec_kwargs.pop("focus_node", None)
)
output_dir = exec_kwargs.pop("output_dir", _DEFAULT_OUTPUT_DIR)
operator_sequence = exec_kwargs.pop("operator_sequence", None)
glyph_sequence = exec_kwargs.pop("glyph_sequence", None)
sequence_context = exec_kwargs.pop("sequence_context", None)
# Get recommendations
recommendations = self.recommend_optimization_strategy(G, operation_type)
baseline_snapshot = (
self._capture_structural_snapshot(G) if capture_snapshots else None
)
validation_report = self._prepare_sequence_validation(
operator_sequence,
glyph_sequence,
sequence_context,
G=G,
node=node_label,
)
if dry_run:
payload = self._build_dry_run_payload(
recommendations,
baseline_snapshot,
validation_report,
operation_type,
seed_value,
node_label,
)
safe_seed = _sanitize_label(seed_value, "unseeded")
safe_node = _sanitize_label(node_label, "global")
snapshot_path, signature = self._persist_dry_run_payload(
payload,
output_dir,
safe_seed,
safe_node,
)
snapshots = (
{
"before": baseline_snapshot,
"after": baseline_snapshot,
}
if baseline_snapshot
else None
)
return {
"dry_run": True,
"snapshot_path": str(snapshot_path),
"signature": signature,
"recommendations": recommendations,
"learning_updated": False,
"telemetry_snapshots": snapshots,
"validation": validation_report,
}
# Apply best strategy
if recommendations.recommended_strategies and self.orchestrator:
best_strategy = recommendations.recommended_strategies[0]
# Map strategy name to OptimizationStrategy enum
strategy_mapping = {
"spectral_methods": OptimizationStrategy.SPECTRAL_FFT,
"vectorized": OptimizationStrategy.NODAL_VECTORIZED,
"cache": OptimizationStrategy.ADELIC_CACHE,
"structural": OptimizationStrategy.STRUCTURAL_MEMO,
"hybrid": OptimizationStrategy.HYBRID,
}
# Find matching strategy
optimization_strategy = OptimizationStrategy.AUTO
for name_part, strategy in strategy_mapping.items():
if name_part in best_strategy.lower():
optimization_strategy = strategy
break
# Conservation pre-check (P5: capture baseline conserved quantities)
conservation_before = None
if HAS_CONSERVATION and G is not None:
try:
conservation_before = capture_conservation_snapshot(G)
except Exception:
conservation_before = None
# Execute optimization
try:
profile = self.orchestrator.analyze_optimization_profile(
G, operation_type
)
result = self.orchestrator.execute_optimization(
G, operation_type, optimization_strategy, **exec_kwargs
)
# Conservation post-check (P5: verify conservation balance)
conservation_result = None
conservation_healthy = True
if (
conservation_before is not None
and HAS_CONSERVATION
and G is not None
):
try:
conservation_after = capture_conservation_snapshot(G)
balance = verify_conservation_balance(
conservation_before, conservation_after
)
lyapunov = compute_lyapunov_derivative(
conservation_before, conservation_after
)
violations = detect_grammar_violations_from_conservation(
balance
)
conservation_result = {
"charge_drift": balance.charge_drift,
"rms_residual": balance.rms_residual,
"lyapunov_stable": lyapunov.is_stable,
"energy_derivative": lyapunov.energy_derivative,
"violations_detected": violations["violations_detected"],
"violation_types": violations.get("violation_types", []),
}
conservation_healthy = not violations["violations_detected"]
except Exception:
pass
# Record experience (P5: includes conservation metrics)
perf_metrics: dict[str, float] = {
"speedup_factor": result.speedup_factor,
"execution_time": result.execution_time,
"memory_used_mb": result.memory_used_mb,
"cache_hits": result.cache_hits,
}
if conservation_result is not None:
perf_metrics["conservation_charge_drift"] = conservation_result[
"charge_drift"
]
perf_metrics["conservation_energy_derivative"] = (
conservation_result["energy_derivative"]
)
perf_metrics["conservation_rms_residual"] = conservation_result[
"rms_residual"
]
experience = OptimizationExperience(
graph_properties={
"nodes": len(G.nodes()) if HAS_NETWORKX and G else 0,
"edges": len(G.edges()) if HAS_NETWORKX and G else 0,
"density": profile.edge_density,
},
operation_type=operation_type,
strategy_used=optimization_strategy.value,
parameters=exec_kwargs,
performance_metrics=perf_metrics,
timestamp=time.time(),
success=result.accuracy_preserved and conservation_healthy,
mathematical_signature=recommendations.mathematical_insights,
)
self.learn_from_experience(experience)
after_snapshot = (
self._capture_structural_snapshot(G) if capture_snapshots else None
)
return {
"optimization_result": result,
"strategy_used": optimization_strategy.value,
"recommendations": recommendations,
"learning_updated": True,
"conservation": conservation_result,
"telemetry_snapshots": (
{
"before": baseline_snapshot,
"after": after_snapshot,
}
if capture_snapshots
else None
),
"validation": validation_report,
}
except Exception as e:
return {
"error": str(e),
"recommendations": recommendations,
"learning_updated": False,
"telemetry_snapshots": (
{"before": baseline_snapshot} if baseline_snapshot else None
),
"validation": validation_report,
}
else:
return {
"message": "No optimization applied",
"recommendations": recommendations,
"learning_updated": False,
"telemetry_snapshots": (
{"before": baseline_snapshot} if baseline_snapshot else None
),
"validation": validation_report,
}
def export_learned_knowledge(self) -> dict[str, Any]:
"""Export learned optimization knowledge."""
return {
"learned_policies": [
{
"name": p.policy_name,
"objective": p.objective.value,
"conditions": p.conditions,
"actions": p.actions,
"confidence": p.confidence,
"success_rate": p.success_rate,
"average_improvement": p.average_improvement,
"applications": p.applications_count,
}
for p in self.learned_policies
],
"adaptive_configuration": dict(self.adaptive_config),
"performance_statistics": {
"total_attempts": self.optimization_attempts,
"successful_optimizations": self.successful_optimizations,
"success_rate": self.successful_optimizations
/ max(1, self.optimization_attempts),
"experience_count": len(self.experience_history),
},
"mathematical_insights": dict(self.mathematical_insights),
}
def import_learned_knowledge(self, knowledge: dict[str, Any]) -> None:
"""Import previously learned optimization knowledge."""
with self._lock:
# Import policies
if "learned_policies" in knowledge:
imported_policies = []
for p_data in knowledge["learned_policies"]:
policy = OptimizationPolicy(
policy_name=p_data["name"],
objective=OptimizationObjective(p_data["objective"]),
conditions=p_data["conditions"],
actions=p_data["actions"],
confidence=p_data["confidence"],
success_rate=p_data["success_rate"],
average_improvement=p_data["average_improvement"],
applications_count=p_data.get("applications", 0),
)
imported_policies.append(policy)
self.learned_policies.extend(imported_policies)
# Import adaptive configuration
if "adaptive_configuration" in knowledge:
self.adaptive_config.update(knowledge["adaptive_configuration"])
# Import insights
if "mathematical_insights" in knowledge:
self.mathematical_insights.update(knowledge["mathematical_insights"])
def _capture_structural_snapshot(self, G: Any) -> dict[str, Any] | None:
"""Capture graph metrics and unified telemetry for reporting."""
timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
if G is None or not HAS_NETWORKX:
snapshot = {"timestamp": timestamp, "graph": None, "telemetry": None}
else:
num_nodes = len(G.nodes())
num_edges = len(G.edges())
density = (
(2 * num_edges) / (num_nodes * (num_nodes - 1)) if num_nodes > 1 else 0
)
graph_metrics = {
"nodes": num_nodes,
"edges": num_edges,
"density": density,
"is_connected": nx.is_connected(G) if HAS_NETWORKX else False,
}
snapshot = {
"timestamp": timestamp,
"graph": graph_metrics,
"telemetry": None,
}
if HAS_UNIFIED_FIELDS and G is not None:
try:
telemetry = compute_unified_telemetry(G)
snapshot["telemetry"] = _json_safe(telemetry)
except (
Exception
) as exc: # pragma: no cover - telemetry errors are informational
snapshot["telemetry_error"] = str(exc)
if HAS_METRIC_OPERATORS and G is not None and HAS_NETWORKX:
coherence_value: float | None = None
sense_value: float | None = None
try:
coherence_value = (
float(compute_coherence(G)) if compute_coherence else None
)
except Exception:
coherence_value = None
try:
if compute_Si:
si_payload = compute_Si(G, inplace=False)
sense_value = _sense_index_mean(si_payload)
except Exception:
sense_value = None
if coherence_value is not None:
snapshot["coherence"] = coherence_value
if sense_value is not None:
snapshot["sense_index"] = sense_value
return snapshot
def _prepare_sequence_validation(
self,
operator_sequence: Any,
glyph_sequence: Any,
context: dict[str, Any] | None,
*,
G: Any = None,
node: Any = None,
) -> dict[str, Any] | None:
"""Validate operator/glyph sequences when provided.
Performs two layers of validation:
1. **Batch** — full-sequence grammar check via ``validate_sequence()``.
2. **Incremental** — per-step check against the node's live history
via ``validate_sequence_incremental()`` (when *G* and *node* are
available). This catches violations that only become visible in
context (e.g. destabilizer debt from prior operations).
"""
sequence = (
operator_sequence if operator_sequence is not None else glyph_sequence
)
if sequence is None:
return None
if isinstance(sequence, str):
sequence_iterable = [sequence]
else:
sequence_iterable = list(sequence)
tokens = _canonicalize_sequence_tokens(sequence_iterable)
if not tokens:
return None
if context:
outcome = validate_sequence(tokens, context=context)
else:
outcome = validate_sequence(tokens)
summary = _json_safe(getattr(outcome, "summary", {}))
if not outcome.passed:
message = (
summary.get("message") or outcome.message or "grammar validation failed"
)
raise TNFRValueError(
f"Operator sequence failed grammar validation: {message}",
context={
"tokens": tokens,
"summary": summary,
"outcome_message": outcome.message,
},
suggestion="Ensure sequence follows U1-U6 grammar rules.",
)
# Incremental per-node validation (proactive, GAP #4)
incremental_report: list[dict[str, Any]] | None = None
if G is not None and node is not None and node in G.nodes:
try:
from ..operators.grammar_dynamics import validate_sequence_incremental
step_results = validate_sequence_incremental(G, node, tokens)
step_violations = [
{
"step": i,
"token": sr.candidate,
"allowed": sr.allowed,
"violations": [
{
"rule": v.rule,
"message": v.message,
"severity": v.severity,
}
for v in sr.violations
],
"suggested": sr.suggested_alternative,
}
for i, sr in enumerate(step_results)
if sr.violations
]
if step_violations:
incremental_report = step_violations
except Exception:
pass # incremental validation is advisory, never blocks
report: dict[str, Any] = {
"passed": True,
"message": outcome.message,
"summary": summary,
"tokens": _json_safe(list(getattr(outcome, "tokens", tokens))),
"canonical_tokens": _json_safe(
list(getattr(outcome, "canonical_tokens", tokens))
),
}
if incremental_report:
report["incremental_violations"] = incremental_report
return report
def _serialize_recommendations(
self, recommendations: SelfOptimizationResult
) -> dict[str, Any]:
"""Convert recommendation dataclass into JSON-serializable payload."""
return {
"recommended_strategies": list(recommendations.recommended_strategies),
"predicted_speedups": _json_safe(recommendations.predicted_speedups),
"optimization_improvements": _json_safe(
recommendations.optimization_improvements
),
"mathematical_insights": _json_safe(recommendations.mathematical_insights),
"adaptive_configurations": _json_safe(
recommendations.adaptive_configurations
),
"execution_time": recommendations.execution_time,
"learned_policies": [
{
"name": policy.policy_name,
"objective": policy.objective.value,
"conditions": _json_safe(policy.conditions),
"actions": _json_safe(policy.actions),
"confidence": policy.confidence,
"success_rate": policy.success_rate,
"average_improvement": policy.average_improvement,
"applications": policy.applications_count,
}
for policy in recommendations.learned_policies
],
}
def _build_dry_run_payload(
self,
recommendations: SelfOptimizationResult,
baseline_snapshot: dict[str, Any] | None,
validation_report: dict[str, Any] | None,
operation_type: str,
seed_value: Any,
node_label: Any,
) -> dict[str, Any]:
"""Assemble payload for dry-run persistence."""
timestamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
seed_label = _sanitize_label(seed_value, "unseeded")
node_name = _sanitize_label(node_label, "global")
telemetry_snapshots = None
if baseline_snapshot is not None:
telemetry_snapshots = {
"before": baseline_snapshot,
"after": baseline_snapshot,
}
payload = {
"metadata": {
"timestamp": timestamp,
"operation_type": operation_type,
"seed": seed_label,
"node": node_name,
"dry_run": True,
"objective": self.optimization_objective.value,
"learning_strategy": self.learning_strategy.value,
},
"telemetry": telemetry_snapshots,
"recommendations": self._serialize_recommendations(recommendations),
"validation": validation_report or {"status": "not_provided"},
"learning_state": {
"experience_count": len(self.experience_history),
"policy_count": len(self.learned_policies),
"successful_optimizations": self.successful_optimizations,
},
}
return _json_safe(payload)
def _persist_dry_run_payload(
self,
payload: dict[str, Any],
output_dir: Any,
safe_seed: str,
safe_node: str,
) -> tuple[Path, str]:
"""Write payload to disk and emit SHA-256 signature."""
base_dir = Path(output_dir)
target_dir = base_dir / safe_seed
target_dir.mkdir(parents=True, exist_ok=True)
file_path = target_dir / f"{safe_node}.json"
serialized = json.dumps(payload, indent=2, sort_keys=True)
file_path.write_text(serialized, encoding="utf-8")
signature = hashlib.sha256(serialized.encode("utf-8")).hexdigest()
signature_path = file_path.with_suffix(file_path.suffix + ".sha256")
signature_path.write_text(f"{signature} {file_path.name}\n", encoding="utf-8")
return file_path, signature
# Factory functions
def create_self_optimizing_engine(**kwargs) -> TNFRSelfOptimizingEngine:
"""Create self-optimizing engine."""
return TNFRSelfOptimizingEngine(**kwargs)
def auto_optimize_tnfr_computation(
G: Any, operation_type: str = "general", **kwargs
) -> dict[str, Any]:
"""Convenience function for automatic optimization."""
engine = create_self_optimizing_engine()
return engine.optimize_automatically(G, operation_type, **kwargs)