Skip to content

Commit 7cce68c

Browse files
Copilotfermga
andcommitted
Implement extended SHA metrics with helper functions and comprehensive tests
Co-authored-by: fermga <203334638+fermga@users.noreply.github.com>
1 parent c02d8f4 commit 7cce68c

File tree

2 files changed

+584
-27
lines changed

2 files changed

+584
-27
lines changed

src/tnfr/operators/metrics.py

Lines changed: 196 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -832,11 +832,150 @@ def resonance_metrics(
832832
}
833833

834834

835+
def _compute_epi_variance(G: TNFRGraph, node: NodeId) -> float:
836+
"""Compute EPI variance during silence period.
837+
838+
Measures the standard deviation of EPI values recorded during silence,
839+
validating effective preservation (variance ≈ 0).
840+
841+
Parameters
842+
----------
843+
G : TNFRGraph
844+
Graph containing the node
845+
node : NodeId
846+
Node to compute variance for
847+
848+
Returns
849+
-------
850+
float
851+
Standard deviation of EPI during silence period
852+
"""
853+
import numpy as np
854+
855+
epi_history = G.nodes[node].get("epi_history_during_silence", [])
856+
if len(epi_history) < 2:
857+
return 0.0
858+
return float(np.std(epi_history))
859+
860+
861+
def _compute_preservation_integrity(preserved_epi: float, epi_after: float) -> float:
862+
"""Compute preservation integrity ratio.
863+
864+
Measures structural preservation quality as:
865+
integrity = 1 - |EPI_after - EPI_preserved| / EPI_preserved
866+
867+
Interpretation:
868+
- integrity = 1.0: Perfect preservation
869+
- integrity < 0.95: Significant degradation
870+
- integrity < 0.8: Preservation failure
871+
872+
Parameters
873+
----------
874+
preserved_epi : float
875+
EPI value that was preserved at silence start
876+
epi_after : float
877+
Current EPI value
878+
879+
Returns
880+
-------
881+
float
882+
Preservation integrity in [0, 1]
883+
"""
884+
if preserved_epi == 0:
885+
return 1.0 if epi_after == 0 else 0.0
886+
887+
integrity = 1.0 - abs(epi_after - preserved_epi) / abs(preserved_epi)
888+
return max(0.0, integrity)
889+
890+
891+
def _compute_reactivation_readiness(G: TNFRGraph, node: NodeId) -> float:
892+
"""Compute readiness score for reactivation from silence.
893+
894+
Evaluates if the node can reactivate effectively based on:
895+
- νf residual (must be recoverable)
896+
- EPI preserved (must be coherent)
897+
- Silence duration (not excessive)
898+
- Network connectivity (active neighbors)
899+
900+
Score in [0, 1]:
901+
- 1.0: Fully ready to reactivate
902+
- 0.5-0.8: Moderate readiness
903+
- < 0.3: Risky reactivation
904+
905+
Parameters
906+
----------
907+
G : TNFRGraph
908+
Graph containing the node
909+
node : NodeId
910+
Node to compute readiness for
911+
912+
Returns
913+
-------
914+
float
915+
Reactivation readiness score in [0, 1]
916+
"""
917+
vf = _get_node_attr(G, node, ALIAS_VF)
918+
epi = _get_node_attr(G, node, ALIAS_EPI)
919+
duration = G.nodes[node].get("silence_duration", 0.0)
920+
921+
# Count active neighbors
922+
active_neighbors = 0
923+
if G.has_node(node):
924+
for n in G.neighbors(node):
925+
if _get_node_attr(G, n, ALIAS_VF) > 0.1:
926+
active_neighbors += 1
927+
928+
# Scoring components
929+
vf_score = min(vf / 0.5, 1.0) # νf recoverable
930+
epi_score = min(epi / 0.3, 1.0) # EPI coherent
931+
duration_score = 1.0 / (1.0 + duration * 0.1) # Penalize long silence
932+
network_score = min(active_neighbors / 3.0, 1.0) # Network support
933+
934+
return (vf_score + epi_score + duration_score + network_score) / 4.0
935+
936+
937+
def _estimate_time_to_collapse(G: TNFRGraph, node: NodeId) -> float:
938+
"""Estimate time until nodal collapse during silence.
939+
940+
Estimates how long silence can be maintained before structural collapse
941+
based on observed drift rate or default degradation model.
942+
943+
Model:
944+
t_collapse ≈ EPI_preserved / |DRIFT_RATE|
945+
946+
Parameters
947+
----------
948+
G : TNFRGraph
949+
Graph containing the node
950+
node : NodeId
951+
Node to estimate collapse time for
952+
953+
Returns
954+
-------
955+
float
956+
Estimated time steps until collapse (inf if no degradation)
957+
"""
958+
preserved_epi = G.nodes[node].get("preserved_epi", 0.0)
959+
drift_rate = G.nodes[node].get("epi_drift_rate", 0.0)
960+
961+
if abs(drift_rate) < 1e-10:
962+
# No observed degradation - return large value
963+
return float("inf")
964+
965+
if preserved_epi <= 0:
966+
# Already at or below collapse threshold
967+
return 0.0
968+
969+
# Estimate time until EPI reaches zero
970+
return abs(preserved_epi / drift_rate)
971+
972+
835973
def silence_metrics(
836974
G: TNFRGraph, node: NodeId, vf_before: float, epi_before: float
837975
) -> dict[str, Any]:
838-
"""SHA - Silence metrics: νf reduction, EPI preservation, and latency tracking.
976+
"""SHA - Silence metrics: νf reduction, EPI preservation, duration tracking.
839977
978+
Extended metrics for deep analysis of structural preservation effectiveness.
840979
Collects silence-specific metrics that reflect canonical SHA effects including
841980
latency state management as specified in TNFR.pdf §2.3.10.
842981
@@ -855,46 +994,76 @@ def silence_metrics(
855994
-------
856995
dict
857996
Silence-specific metrics including:
858-
- Core metrics: vf_reduction, epi_preservation
859-
- Latency state: latent flag, silence_duration
860-
- Integrity metrics: preservation_integrity, epi_variance
997+
998+
**Core metrics (existing):**
999+
1000+
- operator: "Silence"
1001+
- glyph: "SHA"
1002+
- vf_reduction: Absolute reduction in νf
1003+
- vf_final: Post-silence νf value
1004+
- epi_preservation: Absolute EPI change (should be ≈ 0)
1005+
- epi_final: Post-silence EPI value
1006+
- is_silent: Boolean indicating silent state (νf < 0.1)
1007+
1008+
**Latency state tracking:**
1009+
1010+
- latent: Boolean latency flag
1011+
- silence_duration: Time in silence state (steps or structural time)
1012+
1013+
**Extended metrics (NEW):**
1014+
1015+
- epi_variance: Standard deviation of EPI during silence
1016+
- preservation_integrity: Quality metric [0, 1] for preservation
1017+
- reactivation_readiness: Readiness score [0, 1] for reactivation
1018+
- time_to_collapse: Estimated time until nodal collapse
1019+
1020+
Notes
1021+
-----
1022+
Extended metrics enable:
1023+
- Detection of excessive silence (collapse risk)
1024+
- Validation of preservation quality
1025+
- Analysis of consolidation patterns (memory, learning)
1026+
- Strategic pause effectiveness (biomedical, cognitive, social domains)
1027+
1028+
See Also
1029+
--------
1030+
_compute_epi_variance : EPI variance computation
1031+
_compute_preservation_integrity : Preservation quality metric
1032+
_compute_reactivation_readiness : Reactivation readiness score
1033+
_estimate_time_to_collapse : Collapse time estimation
8611034
"""
8621035
vf_after = _get_node_attr(G, node, ALIAS_VF)
8631036
epi_after = _get_node_attr(G, node, ALIAS_EPI)
1037+
preserved_epi = G.nodes[node].get("preserved_epi")
8641038

865-
# Basic SHA metrics
866-
metrics = {
1039+
# Core metrics (existing)
1040+
core = {
8671041
"operator": "Silence",
8681042
"glyph": "SHA",
8691043
"vf_reduction": vf_before - vf_after,
8701044
"vf_final": vf_after,
8711045
"epi_preservation": abs(epi_after - epi_before),
8721046
"epi_final": epi_after,
873-
"is_silent": vf_after < 0.1, # Configurable threshold
1047+
"is_silent": vf_after < 0.1,
8741048
}
8751049

8761050
# Latency state tracking metrics
877-
metrics["latent"] = G.nodes[node].get("latent", False)
878-
metrics["silence_duration"] = G.nodes[node].get("silence_duration", 0.0)
879-
880-
# Preservation integrity: measures EPI variance during silence
881-
preserved_epi = G.nodes[node].get("preserved_epi")
882-
if preserved_epi is not None:
883-
preservation_integrity = abs(epi_after - preserved_epi) / max(
884-
abs(preserved_epi), 1e-10
885-
)
886-
metrics["preservation_integrity"] = preservation_integrity
887-
else:
888-
metrics["preservation_integrity"] = 0.0
889-
890-
# EPI variance during silence (relative to preserved value)
891-
if preserved_epi is not None:
892-
epi_variance = abs(epi_after - preserved_epi)
893-
metrics["epi_variance_during_silence"] = epi_variance
894-
else:
895-
metrics["epi_variance_during_silence"] = abs(epi_after - epi_before)
1051+
core["latent"] = G.nodes[node].get("latent", False)
1052+
core["silence_duration"] = G.nodes[node].get("silence_duration", 0.0)
1053+
1054+
# Extended metrics (new)
1055+
extended = {
1056+
"epi_variance": _compute_epi_variance(G, node),
1057+
"preservation_integrity": (
1058+
_compute_preservation_integrity(preserved_epi, epi_after)
1059+
if preserved_epi is not None
1060+
else 1.0 - abs(epi_after - epi_before)
1061+
),
1062+
"reactivation_readiness": _compute_reactivation_readiness(G, node),
1063+
"time_to_collapse": _estimate_time_to_collapse(G, node),
1064+
}
8961065

897-
return metrics
1066+
return {**core, **extended}
8981067

8991068

9001069
def expansion_metrics(

0 commit comments

Comments
 (0)