Typorithm · medium

Catalog analytics with identity, chronology and honest coverage

pythonpandasnumerical-computingindependent-practice
Saved on this device
Full lesson · 4 learning steps Open any exercise for focused practice
1

Step 1 of 4 · intro card

Catalog analytics with identity, chronology and honest coverage

0%

Concept and contract

Build a synthetic technical-role report without stale duplicates or invented measurements.

A trustworthy catalog report combines several small contracts rather than one impressive expression. Define a role identity, select its latest observation, validate company lookup cardinality, choose the geographic and technical population, then summarize observed measurements with coverage. The order matters: filtering before selecting the latest observation can resurrect a stale eligible record after a role moves to a different country or becomes ineligible.

This capstone uses synthetic observations with explicit sequence metadata and an owned company dimension. It returns one grouped report, not a feed importer or a production data migration. Historical observations remain untouched outside this derived view.

Mechanism and worked example

Validate nonblank role and company IDs, nonnegative integer sequences and boolean technical flags.

Validate nonblank role and company IDs, nonnegative integer sequences and boolean technical flags. Require unique ID-sequence observation pairs and a unique company dimension. Parse updated timestamps as UTC, sort observations by ID, time and sequence and retain the newest per ID. Perform a validated many-to-one left lookup with an indicator, then require a matched company, a technical role and a country in UA, IL, US or CA.

Aggregate only that retained population: roles counts every eligible role, measured counts known scores and mean_score averages observed scores. Include company count separately and sort countries. A role with a missing score remains a role rather than disappearing from the report.

Complete worked example

python

import pandas as pd

def inspect_example(rows, companies):
    if any(type(r.get("id")) is not str or not r["id"].strip() or type(r.get("company_id")) is not str or not r["company_id"].strip() or type(r.get("seq")) is not int or r["seq"] < 0 or type(r.get("technical")) is not bool for r in rows):
        raise ValueError("invalid role observation")
    frame = pd.DataFrame(rows, columns=["id", "company_id", "country", "technical", "updated", "seq", "score"])
    frame = frame.astype({"id": "string", "company_id": "string", "country": "string", "technical": "boolean", "score": "Float64"})
    dimension = pd.DataFrame(companies, columns=["company_id"]).astype("string")
    if frame.duplicated(["id", "seq"]).any() or dimension["company_id"].isna().any() or dimension["company_id"].duplicated().any():
        raise ValueError("ambiguous identity")
    frame["country"] = frame["country"].str.strip().str.upper()
    frame["updated"] = pd.to_datetime(frame["updated"], format="ISO8601", utc=True, errors="raise")
    latest = frame.sort_values(["id", "updated", "seq"], kind="stable").drop_duplicates("id", keep="last")
    joined = latest.merge(dimension, on="company_id", how="left", validate="many_to_one", indicator=True)
    eligible = joined["_merge"].eq("both") & joined["technical"] & joined["country"].isin(["UA", "IL", "US", "CA"])
    selected = joined.loc[eligible.fillna(False)]
    report = selected.groupby("country").agg(roles=("id", "size"), companies=("company_id", "nunique"), measured=("score", "count"), mean_score=("score", "mean")).reset_index()
    return report.astype({"country": "string", "roles": "Int64", "companies": "Int64", "measured": "Int64", "mean_score": "Float64"}).sort_values("country").reset_index(drop=True)
Reasoning
Validate nonblank role and company IDs, nonnegative integer sequences and boolean technical flags. Require unique ID-sequence observation pairs and a unique company dimension. Parse updated timestamps as UTC, sort observations by ID, time and sequence and retain the newest per ID. Perform a validated many-to-one left lookup with an indicator, then require a matched company, a technical role and a country in UA, IL, US or CA. Aggregate only that retained population: roles counts every eligible role, measured counts known scores and mean_score averages observed scores. Include company count separately and sort countries. A role with a missing score remains a role rather than disappearing from the report.
Open in local Python playground ↗

Focused decision: context is in the complete example

python

latest = frame.sort_values(["id", "updated", "seq"], kind="stable")
latest = latest.drop_duplicates("id", keep="last")
joined = latest.merge(dimension, on="company_id", how="left", validate="many_to_one", indicator=True)
eligible = joined["_merge"].eq("both") & joined["technical"]
eligible &= joined["country"].isin(["UA", "IL", "US", "CA"])
Focused decision
Type this short table decision after reading its complete context. The complete imports, table construction, validation, and result checks remain in the worked example. Typing records practice; only the downloadable computer lab executes Python.

Counterexamples and common mistakes

The controlled mistake filters technical rows before selecting each role's latest observation.

The controlled mistake filters technical rows before selecting each role's latest observation. If an older observation is technical but the latest is not, the stale observation re-enters the report. Another defect fills missing scores with zero before aggregation, overstating coverage and changing means. A third joins an unvalidated company dimension, multiplying role counts. Each mistake can produce a polished but inaccurate dashboard.

Country here means the normalized explicit observation field, not company headquarters. The sample does not infer remote-country eligibility, cross-border legal access or the full market from a company label. Unknown or unmatched records require a separate quality view rather than invented classifications.

Controlled mistake: inspect the failing table contract

python

import pandas as pd

def inspect_example(rows, companies):
    if any(type(r.get("id")) is not str or not r["id"].strip() or type(r.get("company_id")) is not str or not r["company_id"].strip() or type(r.get("seq")) is not int or r["seq"] < 0 or type(r.get("technical")) is not bool for r in rows):
        raise ValueError("invalid role observation")
    frame = pd.DataFrame(rows, columns=["id", "company_id", "country", "technical", "updated", "seq", "score"])
    frame = frame.astype({"id": "string", "company_id": "string", "country": "string", "technical": "boolean", "score": "Float64"})
    dimension = pd.DataFrame(companies, columns=["company_id"]).astype("string")
    if frame.duplicated(["id", "seq"]).any() or dimension["company_id"].isna().any() or dimension["company_id"].duplicated().any():
        raise ValueError("ambiguous identity")
    frame["country"] = frame["country"].str.strip().str.upper()
    frame["updated"] = pd.to_datetime(frame["updated"], format="ISO8601", utc=True, errors="raise")
    frame = frame.loc[frame["technical"]]
    latest = frame.sort_values(["id", "updated", "seq"], kind="stable").drop_duplicates("id", keep="last")
    joined = latest.merge(dimension, on="company_id", how="left", validate="many_to_one", indicator=True)
    eligible = joined["_merge"].eq("both") & joined["technical"] & joined["country"].isin(["UA", "IL", "US", "CA"])
    selected = joined.loc[eligible.fillna(False)]
    report = selected.groupby("country").agg(roles=("id", "size"), companies=("company_id", "nunique"), measured=("score", "count"), mean_score=("score", "mean")).reset_index()
    return report.astype({"country": "string", "roles": "Int64", "companies": "Int64", "measured": "Int64", "mean_score": "Float64"}).sort_values("country").reset_index(drop=True)
Reasoning
The controlled mistake filters technical rows before selecting each role's latest observation. If an older observation is technical but the latest is not, the stale observation re-enters the report. Another defect fills missing scores with zero before aggregation, overstating coverage and changing means. A third joins an unvalidated company dimension, multiplying role counts. Each mistake can produce a polished but inaccurate dashboard. Country here means the normalized explicit observation field, not company headquarters. The sample does not infer remote-country eligibility, cross-border legal access or the full market from a company label. Unknown or unmatched records require a separate quality view rather than invented classifications.
Open in local Python playground ↗

Independent practice

Restrict the visible report to UA and IL, retaining all other pipeline decisions.

Restrict the visible report to UA and IL, retaining all other pipeline decisions. Change only the country allowlist after latest-observation selection and validated enrichment. Do not remove historical US or CA observations from the input, weaken the company relationship check or change the score denominator.

Predict an older technical UA observation followed by a newer nontechnical one: it stays excluded under both contracts. Predict a latest technical US role: it is included in the worked report and excluded by the practice policy. This separates the intended country change from the stale-observation defect.

Step 1: Report UA and IL only without changing latest-role selection or score coverage. Step 2: Retain the working imports, input validation and table schemas. Change the stated policy, run the supplied practice cases, then explain the newly accepted or rejected boundary.

Practice starter: solve before reading the solution

python

import pandas as pd

def inspect_example(rows, companies):
    if any(type(r.get("id")) is not str or not r["id"].strip() or type(r.get("company_id")) is not str or not r["company_id"].strip() or type(r.get("seq")) is not int or r["seq"] < 0 or type(r.get("technical")) is not bool for r in rows):
        raise ValueError("invalid role observation")
    frame = pd.DataFrame(rows, columns=["id", "company_id", "country", "technical", "updated", "seq", "score"])
    frame = frame.astype({"id": "string", "company_id": "string", "country": "string", "technical": "boolean", "score": "Float64"})
    dimension = pd.DataFrame(companies, columns=["company_id"]).astype("string")
    if frame.duplicated(["id", "seq"]).any() or dimension["company_id"].isna().any() or dimension["company_id"].duplicated().any():
        raise ValueError("ambiguous identity")
    frame["country"] = frame["country"].str.strip().str.upper()
    frame["updated"] = pd.to_datetime(frame["updated"], format="ISO8601", utc=True, errors="raise")
    latest = frame.sort_values(["id", "updated", "seq"], kind="stable").drop_duplicates("id", keep="last")
    joined = latest.merge(dimension, on="company_id", how="left", validate="many_to_one", indicator=True)
    eligible = joined["_merge"].eq("both") & joined["technical"] & joined["country"].isin(["UA", "IL", "US", "CA"])
    selected = joined.loc[eligible.fillna(False)]
    report = selected.groupby("country").agg(roles=("id", "size"), companies=("company_id", "nunique"), measured=("score", "count"), mean_score=("score", "mean")).reset_index()
    return report.astype({"country": "string", "roles": "Int64", "companies": "Int64", "measured": "Int64", "mean_score": "Float64"}).sort_values("country").reset_index(drop=True)

# Independent task: Report UA and IL only without changing latest-role selection or score coverage.
Reasoning
Report UA and IL only without changing latest-role selection or score coverage.
Open in local Python playground ↗
  • Hint 1: Find the one field or branch that owns this policy.
  • Hint 2: Compare the first changed acceptance case with the retained baseline.

Solution, tests and trade-offs

The solution narrows the allowlist while preserving chronological deduplication, company matching, technical eligibility and coverage metrics.

The solution narrows the allowlist while preserving chronological deduplication, company matching, technical eligibility and coverage metrics. Tests include a stale eligible observation, equal-time sequence ordering, a missing score, an unmatched company and a non-target country. Invalid role identity or duplicate company keys remain errors. Exact expected country reports are independent fixture calculations.

The capstone also returns a typed empty report when nothing qualifies. Counts represent the fixture's selected roles and distinct matched companies, not an exhaustive market estimate. For a real dashboard, retain excluded counts, freshness and source provenance alongside the published summary so changes can be reconciled after an import.

Practice solution and acceptance checks

python

import pandas as pd

def inspect_example(rows, companies):
    if any(type(r.get("id")) is not str or not r["id"].strip() or type(r.get("company_id")) is not str or not r["company_id"].strip() or type(r.get("seq")) is not int or r["seq"] < 0 or type(r.get("technical")) is not bool for r in rows):
        raise ValueError("invalid role observation")
    frame = pd.DataFrame(rows, columns=["id", "company_id", "country", "technical", "updated", "seq", "score"])
    frame = frame.astype({"id": "string", "company_id": "string", "country": "string", "technical": "boolean", "score": "Float64"})
    dimension = pd.DataFrame(companies, columns=["company_id"]).astype("string")
    if frame.duplicated(["id", "seq"]).any() or dimension["company_id"].isna().any() or dimension["company_id"].duplicated().any():
        raise ValueError("ambiguous identity")
    frame["country"] = frame["country"].str.strip().str.upper()
    frame["updated"] = pd.to_datetime(frame["updated"], format="ISO8601", utc=True, errors="raise")
    latest = frame.sort_values(["id", "updated", "seq"], kind="stable").drop_duplicates("id", keep="last")
    joined = latest.merge(dimension, on="company_id", how="left", validate="many_to_one", indicator=True)
    eligible = joined["_merge"].eq("both") & joined["technical"] & joined["country"].isin(["UA", "IL"])
    selected = joined.loc[eligible.fillna(False)]
    report = selected.groupby("country").agg(roles=("id", "size"), companies=("company_id", "nunique"), measured=("score", "count"), mean_score=("score", "mean")).reset_index()
    return report.astype({"country": "string", "roles": "Int64", "companies": "Int64", "measured": "Int64", "mean_score": "Float64"}).sort_values("country").reset_index(drop=True)
Reasoning
The solution narrows the allowlist while preserving chronological deduplication, company matching, technical eligibility and coverage metrics. Tests include a stale eligible observation, equal-time sequence ordering, a missing score, an unmatched company and a non-target country. Invalid role identity or duplicate company keys remain errors. Exact expected country reports are independent fixture calculations. The capstone also returns a typed empty report when nothing qualifies. Counts represent the fixture's selected roles and distinct matched companies, not an exhaustive market estimate. For a real dashboard, retain excluded counts, freshness and source provenance alongside the published summary so changes can be reconciled after an import.
Open in local Python playground ↗

Interview reasoning and next step

Walk through the pipeline in order and explain what could go wrong if two neighboring stages were reversed.

Walk through the pipeline in order and explain what could go wrong if two neighboring stages were reversed. Distinguish role identity from observation identity, eligibility from measurement availability and a matched company from a merely plausible label. Explain why counts and means need both a clear population and source freshness before being used for decisions.

On the PC, type the focused latest-observation and eligibility decisions, then reconstruct the aggregate denominator from memory. In the app, answer the same reasoning questions and save reading progress. The downloadable lab executes the complete pipeline. Continue with pytest for stronger failure detection, followed by deeper Python data access, statistics and ML with explicit leakage boundaries.

Optional stretch — after the core task works: Add one independently reasoned boundary case and explain its expected value before running it.

Computer practice: https://applydjinn.com/typorithm/pandas-lab-20261106.zip. Install pinned dependencies in an isolated Python environment. Run verify.py, then change the working exercise baseline and run --practice pd-capstone. The real pandas tables, exact schemas, unchanged inputs, CSV parser options and owned in-memory SQL reconciliation use owned fixtures only. No production database, feed, account or paid service is used. This lab executes Python; the PC typing trainer and native reading progress do not execute it.

Build a synthetic technical-role report without stale duplicates or invented measurements.