Skip to content

Merger

merger

merge set of SecurityProviderRecords.

SecurityMerger(matcher: SecurityMatcher | None = None)

Merge SecurityProviderRecords into Security objects.

Source code in src/equities_classifier/matching/merger.py 
def __init__(
    self,
    matcher: SecurityMatcher | None = None,
) -> None:
    """Initialize SecurityMerger object."""

    self._matcher = matcher or SecurityMatcher()

create_securities(provider_records: Sequence[SecurityProviderRecord]) -> list[Security]

Create Security objects from provider records.

Source code in src/equities_classifier/matching/merger.py 
def create_securities(
    self,
    provider_records: Sequence[SecurityProviderRecord],
) -> list[Security]:
    """Create Security objects from provider records."""

    return [
        self._create_security(provider_record)
        for provider_record in provider_records
    ]

merge_enrich_multiple(left: Sequence[Security], right: Sequence[SecurityProviderRecord]) -> list[Security]

Merge provider records into existing Security objects.

Source code in src/equities_classifier/matching/merger.py 
def merge_enrich_multiple(
    self,
    left: Sequence[Security],
    right: Sequence[SecurityProviderRecord],
) -> list[Security]:
    """Merge provider records into existing Security objects."""

    for security in left:
        security = self.merge_enrich_single(security, right)

    return list(left)

merge_enrich_single(security: Security, right: Sequence[SecurityProviderRecord]) -> Security

Merge provider records into existing Security record.

Source code in src/equities_classifier/matching/merger.py 
def merge_enrich_single(
    self,
    security: Security,
    right: Sequence[SecurityProviderRecord],
) -> Security:
    """Merge provider records into existing Security record."""

    matches = [
        provider_record
        for provider_record in right
        if self._matcher.match(security, provider_record).matched
    ]

    if len(matches) >= 1:

        if len(matches) > 1:
            MatchingHelper.other_error_with_message(
                f"Multiple matching provider records for "
                f"{security.name!r} / {security.ticker!r}, {len(matches)} matches."
            )

        provider_record = matches[0]
        self._merge_identifiers(security, provider_record)
        self._merge_provider_attributes(security, provider_record)

    return security