Skip to content

Merger

merger

merge set of SecurityProviderRecords.

JoinType

Security provider record join type.

INNER = 'inner' class-attribute instance-attribute

LEFT = 'left' class-attribute instance-attribute

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], *, join_type: JoinType = JoinType.LEFT) -> 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],
    *,
    join_type: JoinType = JoinType.LEFT,
) -> 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], *, join_type: JoinType = JoinType.LEFT) -> 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],
    *,
    join_type: JoinType = JoinType.LEFT,
) -> Security:
    """Merge provider records into existing Security record."""

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

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

    provider_record = matches[0]

    self._merge_identifiers(security, provider_record,)
    self._merge_provider_attributes(security, provider_record,)

    return security