• 9 min read
  • 1 code sample

Handling missing HTS codes in ETL pipelines

The exact failure this page solves: an ETL batch ingests commercial invoices, EDI 810/856 transmissions, or supplier feeds and one or more line items arrive with a null, empty-string, or structurally malformed Harmonized Tariff Schedule code. Left unhandled, the batch either aborts mid-run — stranding every downstream line item behind a single bad SKU — or silently coerces the gap into a default classification, which corrupts duty math and violates CBP reasonable-care obligations under 19 CFR 141.86. Neither outcome is acceptable in a production clearance environment. The correct behavior is deterministic: detect the gap during ingestion, hold the affected record without blocking the rest of the batch, attach a provisional conservative rate, and emit an immutable audit event before the line item ever reaches the Duty Formula Calculation Frameworks. This page gives you a single, runnable resolver that enforces that contract, plus the verification checklist and the encoding, temporal, and type-coercion gotchas that break naive implementations.

Prerequisites

This solution assumes a specific upstream pipeline state and toolchain. Confirm each before applying it:

  • Python 3.10+ — the resolver uses structural typing (dataclass(frozen=True)), match-friendly enums, and Decimal throughout. Never use float for money or duty rates; ISO 4217 minor-unit rounding is not representable in binary floating point.
  • An active tariff snapshot in memory or Redis. The resolver reads a point-in-time HTS snapshot keyed by digit-only code. That snapshot is produced upstream by Tariff Update Ingestion Pipelines and must expose effective_date, expiry_date, and general_rate per row, following the bitemporal contract defined in HTS Schedule Database Design. If your snapshot lacks validity windows, superseded-rate detection below silently no-ops.
  • A staged, read-committed source table. Line items must be materialized before resolution so the resolver is a pure function of (item, snapshot, as_of). Do not resolve while a tariff swap is in flight.
  • A dead-letter/quarantine table and a broker-review queue already provisioned. This page routes to them; it does not create them.
  • Structured logging configured (stdlib logging with the %(asctime)s | %(levelname)s | %(name)s | %(message)s format, or structlog) so every fallback event is greppable for audit.

Implementation

A single resolver owns the entire missing-code decision tree: format validation, snapshot lookup, temporal effectiveness, conservative fallback, and broker escalation. It never guesses a filable classification and never lets a gap reach duty computation unaudited.

import logging
import hashlib
from dataclasses import dataclass, field
from datetime import date, datetime
from enum import Enum
from typing import Optional, Dict, Any
from decimal import Decimal, ROUND_HALF_UP

# Structured logging so every fallback event is greppable for CBP audit.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    handlers=[logging.StreamHandler()],
)
logger = logging.getLogger("hts_etl_resolver")


class HtsStatus(Enum):
    VALID = "VALID"
    FALLBACK_HEURISTIC = "FALLBACK_HEURISTIC"
    BROKER_REVIEW = "BROKER_REVIEW"


@dataclass(frozen=True)
class LineItem:
    sku: str
    description: str
    quantity: int
    unit_price: Decimal            # Decimal, never float — ISO 4217 minor units
    hts_code: Optional[str] = None  # None, "", or malformed are all "missing"
    origin_country: Optional[str] = None


@dataclass
class ClassificationResult:
    line_item: LineItem
    resolved_hts: str
    status: HtsStatus
    confidence_score: float
    duty_rate: Decimal
    audit_hash: str
    timestamp: datetime = field(default_factory=lambda: datetime.now(tz=None))


class HtsResolver:
    """Deterministic missing-HTS resolver. Pure function of (item, snapshot, as_of)."""

    def __init__(self, active_snapshot: Dict[str, Dict[str, Any]]):
        self._snapshot = active_snapshot
        self._placeholder = "9999999999"          # provisional bucket, never filable
        self._max_duty_rate = Decimal("0.2500")   # conservative ceiling until reviewed

    def _digits(self, code: Optional[str]) -> Optional[str]:
        # Treat None AND empty string AND non-digit as "missing" — the three
        # shapes malformed upstream feeds actually deliver.
        if not code:
            return None
        stripped = code.strip().replace(".", "")
        return stripped if stripped.isdigit() and len(stripped) in (6, 8, 10) else None

    def _is_effective(self, row: Dict[str, Any], as_of: date) -> bool:
        eff, exp = row.get("effective_date"), row.get("expiry_date")
        if eff and as_of < eff:
            return False
        if exp and as_of > exp:
            return False
        return True

    def _audit(self, item: LineItem) -> str:
        payload = f"{item.sku}|{item.description}|{item.hts_code}"
        return hashlib.sha256(payload.encode("utf-8")).hexdigest()

    def resolve(self, item: LineItem, as_of: Optional[date] = None) -> ClassificationResult:
        entry_date = as_of or date.today()
        audit_hash = self._audit(item)
        digits = self._digits(item.hts_code)

        # 1. Structurally valid code that exists in the active snapshot.
        if digits is not None and digits in self._snapshot:
            row = self._snapshot[digits]
            if not self._is_effective(row, entry_date):
                # Code exists historically but is not active on the entry date —
                # never apply a superseded rate automatically (19 USC 1504).
                logger.warning(
                    "Superseded HTS %s for SKU %s; routing to broker review.",
                    digits, item.sku,
                )
                return ClassificationResult(
                    item, digits, HtsStatus.BROKER_REVIEW, 0.0,
                    self._max_duty_rate, audit_hash,
                )
            rate = Decimal(str(row.get("general_rate", "0")))
            logger.info("HTS resolved: %s -> SKU %s @ %s", digits, item.sku, rate)
            return ClassificationResult(
                item, digits, HtsStatus.VALID, 1.0, rate, audit_hash,
            )

        # 2. Missing/malformed/unknown code -> conservative provisional hold.
        logger.warning(
            "Missing or invalid HTS for SKU %s (raw=%r); provisional %s @ max rate.",
            item.sku, item.hts_code, self._placeholder,
        )
        return ClassificationResult(
            item, self._placeholder, HtsStatus.FALLBACK_HEURISTIC, 0.0,
            self._max_duty_rate, audit_hash,
        )


def calculate_duty(result: ClassificationResult) -> Decimal:
    """Ad valorem duty with CBP half-up rounding. Blocks on unverified holds."""
    if result.status is not HtsStatus.VALID:
        raise ValueError(
            f"Duty blocked: {result.line_item.sku} is {result.status.value}; "
            "resolve classification before computation."
        )
    line_value = result.line_item.quantity * result.line_item.unit_price
    return (line_value * result.duty_rate).quantize(
        Decimal("0.01"), rounding=ROUND_HALF_UP
    )

The critical contract is in calculate_duty: it raises on any status other than VALID. A provisional 9999999999 placeholder or a superseded code can never leak into a filed duty figure — it forces the batch to surface the hold to a broker first.

The resolve() decision tree for a single line item A LineItem enters the resolver. First gate: is the HTS code structurally valid (6, 8, or 10 digits after stripping) AND present in the active snapshot? If no, the record returns FALLBACK_HEURISTIC — placeholder 9999999999, confidence 0.0, at the conservative maximum rate — and is held. If yes, a second gate checks whether the code is effective on the entry date. A superseded code returns BROKER_REVIEW — confidence 0.0, max rate — and routes to the review queue. Only an in-window match returns VALID at confidence 1.0 with the snapshot general rate, and only VALID is allowed to reach calculate_duty; the two hold states raise a ValueError instead. LineItem sku · hts_code · as_of digits in (6,8,10) AND in snapshot? no · missing/malformed yes effective on entry date? no · superseded yes FALLBACK_HEURISTIC hts 9999999999 confidence 0.0 rate = max 0.2500 BROKER_REVIEW keeps raw code confidence 0.0 rate = max 0.2500 VALID confidence 1.0 rate = snapshot general_rate calculate_duty() held · raises ValueError held · raises ValueError
Only a structurally valid, in-snapshot, in-window code reaches VALID; both hold states carry the conservative 0.2500 ceiling and confidence 0.0, and calculate_duty() raises on anything but VALID.

Verification steps

Run these checks against a representative batch before trusting the resolver in production:

  1. Null / empty / malformed parity. Feed three items with hts_code=None, "", and "85.17" (too short). All three must return FALLBACK_HEURISTIC with confidence_score == 0.0 — confirm your upstream mapper does not convert missing values into the string "None", which would pass isdigit() as False but still needs the same treatment.
  2. Snapshot version pin. Query the active table (SELECT max(effective_date) FROM hts_schedule_snapshot;). The date passed as as_of must fall inside that snapshot’s window; a stale snapshot causes silent rate drift, not an error.
  3. Superseded-code rejection. Inject a line item whose code was valid last year but has an expiry_date before as_of. Assert the result is BROKER_REVIEW, not VALID — this is the single most common regression when the snapshot loses its validity columns.
  4. Duty arithmetic checksum. For each VALID result, independently compute quantity * unit_price * duty_rate and confirm calculate_duty matches to the cent under ROUND_HALF_UP. A mismatch means a float slipped into unit_price or general_rate.
  5. Block-on-hold gate. Assert calculate_duty raises ValueError for both FALLBACK_HEURISTIC and BROKER_REVIEW. No provisional line item may reach ACE filing.
  6. Audit chain integrity. Match audit_hash across the ingestion log, the resolver output, and the broker manifest for the same SKU. Any divergence indicates payload mutation or a race against an in-flight tariff swap.
  7. Quarantine record count. Compare the count of non-VALID results against rows landed in the broker-review queue. They must be equal; a shortfall means an exception path is swallowing records instead of routing them.

Edge cases & gotchas

  • Character-encoding corruption in descriptions. Latin-1 or Windows-1252 feeds re-decoded as UTF-8 turn accented product names into mojibake, which poisons any keyword heuristic and mis-hashes the audit payload. Normalize to UTF-8 (errors="strict", not "replace") at ingestion and quarantine on decode failure rather than lossily patching bytes.
  • "None" as a literal string. Some upstream serializers stringify Python None before it reaches the pipeline. if not code catches None and "", but "None" is truthy — it survives to _digits, fails isdigit(), and lands in fallback correctly, but your logs will show raw='None'. Treat that as a data-quality bug in the source mapper, not a resolver success.
  • Formatted vs. digit-only codes. Feeds mix 8517.62.0000 and 8517620000. Snapshot keys must be digit-only or lookups miss valid codes and over-route to fallback. _digits normalizes on read; make sure the snapshot loader normalizes on write too.
  • Digit-length ambiguity. A 6-digit international HS code is a valid subheading but is not filable as a 10-digit US HTS. _digits accepts 6/8/10 for lookup, but a 6-digit match should still route to broker review before filing — extend the effective-code branch if your snapshot mixes granularities.
  • Timezone-naive entry dates. date.today() uses the host clock. For a border-crossing entry near midnight UTC, a naive local date can select the wrong tariff window. Pass an explicit, timezone-derived as_of from the entry event, never the server default.
  • Batch abort on one bad row. If you call resolve inside a list comprehension that also calls calculate_duty eagerly, one held SKU raises and kills the whole batch. Resolve the full batch first, then compute duty only over the VALID subset — hold the rest for the broker queue described in Fallback Routing for Unmapped Codes.

Up: Fallback Routing for Unmapped Codes