src.dackar.RCA.cross_pattern.rules

Functions

compute_link_confidence(signal_similarity_score, ...)

Renormalized weighted formula from §4.2.

classify_linkage_precedence(episode_id, doc, ...)

Return the precedence level for an (episode, doc) pair.

compute_time_overlap_hours(episode_window_start, ...)

Compute temporal overlap (in hours) between episode window and doc event window.

classify_support_posture(reinforcing_fm_ids, ...)

Classify the support posture for a candidate based on its linked FM IDs.

classify_linkage_outcome(episodes, candidate_links, ...)

Determine the linkage outcome for one candidate.

apply_stale_confidence_cap(link, cap)

Return a new CrossPatternLink with link_confidence capped at cap.

Module Contents

Renormalized weighted formula from §4.2.

Always present: signal (weight 0.30). Temporal (0.20), FM (0.20), document (0.30) contribute only if non-None. The formula is renormalized over the sum of present weights so that a missing dimension does not silently deflate confidence relative to other links.

The temporal weight is gated on temporal_compatibility_score — the value actually fed into the numerator — NOT on time_overlap_hours. Gating on the raw overlap would count the temporal dimension while contributing a 0.0 score whenever the caller passed a gap-only overlap (formula mode), silently deflating confidence. time_overlap_hours is retained for provenance only.

provenance is mutated to record the contributing terms, their raw weights, and the normalization factor.

Parameters:

  • signal_similarity_score (float)

  • time_overlap_hours (Optional[float])

  • temporal_compatibility_score (Optional[float])

  • fm_alignment_score (Optional[float])

  • document_similarity_score (Optional[float])

  • provenance (Dict[str, Any])

Return type:

float

src.dackar.RCA.cross_pattern.rules.classify_linkage_precedence(episode_id, doc, episode_source_refs)[source]

Return the precedence level for an (episode, doc) pair.

Level 1 — Direct reference: doc.doc_id appears in episode_source_refs. Level 2 — Temporal: event_time_confidence != “absent” (temporal data is

usable). Asset compatibility is scored separately by the linker (asset_match, recorded in provenance) and is deliberately NOT a gate on this level — the name is “temporal”, not “temporal + asset”.

Level 3 — Fallback (semantic/FM): used when neither level 1 nor 2 applies.

Parameters:

  • episode_id (str) – The episode identifier (carried for provenance; not used in logic).

  • doc (src.dackar.RCA.cross_pattern.models.HistoricalDocExtraction) – The HistoricalDocExtraction being evaluated.

  • episode_source_refs (List[str]) – Raw event/document references carried on the episode (e.g. linked CR IDs, work-order IDs, or source_event_id values).

Return type:

int

src.dackar.RCA.cross_pattern.rules.compute_time_overlap_hours(episode_window_start, episode_window_end, doc, max_gap_days)[source]

Compute temporal overlap (in hours) between episode window and doc event window.

Returns:

  • float – Positive → windows overlap; value is overlap duration in hours. Negative → gap between windows (negative float); the caller can apply

    the max_gap_days gate.

  • None – Returned when: - doc.event_time_confidence == "absent" - any required timestamp is missing (episode or doc)

Parameters:
Return type:

Optional[float]

src.dackar.RCA.cross_pattern.rules.classify_support_posture(reinforcing_fm_ids, conflicting_fm_ids)[source]

Classify the support posture for a candidate based on its linked FM IDs.

Parameters:

  • reinforcing_fm_ids (List[str]) – FM IDs from links where doc.fm_id_candidate matches the candidate FM.

  • conflicting_fm_ids (List[str]) – FM IDs from links where doc.fm_id_candidate differs from the candidate FM.

Returns:

  • (support_posture, reinforcement_strength)

  • Rules (in precedence order)

  • 1. Any conflicting links → (“conflicting”, None)

  • 2. Both lists empty → (“unresolved”, None)

  • 3. Exactly one reinforcing → (“reinforcing”, “single”)

  • 4. Two+ reinforcing, all same fm_id → (“reinforcing”, “multiple_consistent”)

  • 5. Two+ reinforcing, mixed fm_ids → (“weakly_supporting”, “mixed”)

  • 6. reinforcing empty, conflicting empty but some docs existed → (“weakly_supporting”, None) – (Note: caller passes empty lists when no links at all; rule 2 applies first.)

Return type:

Tuple[str, Optional[str]]

src.dackar.RCA.cross_pattern.rules.classify_linkage_outcome(episodes, candidate_links, doc_extractions, link_confidence_threshold)[source]

Determine the linkage outcome for one candidate.

Parameters:
Returns:

  • “no_data” – All episodes have index_status “no_episodes_indexed”, or no doc_extractions exist.

  • ”no_match” – Index is populated and episodes were found, but no candidate_links were built.

  • ”below_threshold” – Links exist but none exceeds link_confidence_threshold.

  • ”linked” – At least one link exceeds link_confidence_threshold.

Return type:

str

src.dackar.RCA.cross_pattern.rules.apply_stale_confidence_cap(link, cap)[source]

Return a new CrossPatternLink with link_confidence capped at cap.

Records in provenance whether the cap was applied and at what value. Returns the link unchanged (but as a new copy) when confidence is already below the cap.

Parameters:
Return type:

src.dackar.RCA.cross_pattern.models.CrossPatternLink