Problem Detection and Prioritization

After the matching pipeline completes (Section 2), the system performs a comprehensive analysis to identify data quality issues in both datasets.

Problem Categories

The documentation separates problems into three conceptual categories, but the current implementation only emits stop-level Problem rows. Route entity and route membership problem pages describe planned route diagnostics.

1. Stop Problems (Implemented)

Issues with individual stops (Distance, Attributes, Isolation, Duplicates). See 4.1 Stop Problems.

2. Route Entity Problems (Planned)

Issues with the Route object itself.

• Not yet implemented: route-level Problem rows, metadata mismatch diagnostics, and unmatched route detection.
• See 4.2 Route Entity Problems.

3. Route Membership Problems (Planned)

Issues with the stop-route relationship.

• Goal: Ensure the sequence of stops and their directional roles match.
• See 4.3 Route Membership Problems.

Architecture

Problem detection is built on the same domain models as the matching pipeline. The enriched MatchingOutput (containing MatchRecord, AtlasNode, and OsmNode entities plus duplicate and route-evidence lookups) flows directly into problem detection — no ORM or dictionary conversion is needed.

Polymorphic Predicates

Each problem predicate is a plain function with a polymorphic signature:

def predicate(ctx: ProblemContext, record: MatchRecord | AtlasNode | OsmNode) -> list[ProblemResult]

Predicates use isinstance checks to decide what to evaluate:

• distance_problem, attributes_problem, and contradicts_route_matching_problem only act on MatchRecord (return [] for bare nodes)
• unmatched_problem only acts on bare AtlasNode or OsmNode (returns [] for MatchRecord)
• duplicates_problem acts on all three types

This allows the same STOP_PROBLEM_PIPELINE list to be used for both matched and unmatched records.

Two Invocation Paths

Both paths produce list[ProblemResult]. The current fast importer converts their problem_type and priority fields into plain problem-row payloads before the database transaction.

ProblemResult Value Object

All predicates return ProblemResult, a lightweight frozen dataclass decoupled from SQLAlchemy:

@dataclass(frozen=True)
class ProblemResult:
    problem_type: str        # 'distance', 'attributes', 'contradicts_route_matching', 'unmatched', 'duplicates'
    priority: int            # 1 = P1, 2 = P2, 3 = P3
    has_atlas_duplicate: bool = False
    has_osm_duplicate: bool = False

Priority Levels

All problems use a consistent three-level priority system:

Priority assignment is rule-based and considers factors like:

• Distance thresholds (configurable constants in [context.py](https://github.com/openTdataCH/stop_sync_osm_atlas/blob/main/matching_and_import_db/problem_detection/context.py))
• ATLAS operator abbreviation (business_org_abbr == "SBB" receives the higher distance tolerance)
• Attribute importance (UIC references are more critical than operator names)
• Isolation status (stops with no nearby counterparts are higher priority)

Code Reference