A lightweight, zero-dependency Python client for the CommissionSight API. It mirrors the surface area of the official TypeScript SDK.
CommissionSight ingests carrier commission statements (CSV/XLSX), normalizes them across carriers, and scores every member period-over-period as 🟢 green / 🟡 yellow / 🔴 red with explicit change flags — so you can see new business, commission changes, and attrition at a glance.
- Zero runtime dependencies — just the Python standard library (
urllib). - Typed — response shapes are exported as
TypedDicts; the package shipspy.typed. - Testable — inject a custom transport (or your own
requests/httpx) for tests and non-standard runtimes. - Python 3.8+.
pip install commissionsightfrom commissionsight import CommissionSightClient
cs = CommissionSightClient(
"https://api.commissionsight.com/v1",
token="...", # a per-account API token
)
carriers = cs.list_carriers()
print(carriers["data"]) # [{"id": ..., "name": ..., "slug": ...}, ...]CommissionSightClient(
base_url, # e.g. https://api.commissionsight.com/v1 (trailing slash optional)
token=None, # Bearer token; can also be set later via set_token()
transport=None, # custom transport (method, url, headers, body) -> (status, text)
)Set or rotate the token at any time:
cs.set_token(new_token)
cs.set_token(None) # clear itThe SDK is for server-to-server integrations, authenticated with a per-account API token issued to you by CommissionSight. Every request is sent as Authorization: Bearer <token>.
Uploading a file kicks off an asynchronous ingest job. Poll the job until it's completed, then read the scored results.
import time
# `file` can be a path, bytes, a file object, or a (filename, content) tuple.
res = cs.upload_file(
"statements/aetna-2026-05.csv",
carrier_id="car_123",
period_year=2026,
period_month=5,
webhook_url="https://acme.com/hooks/commissionsight", # optional
idempotency_key="acme-2026-05-aetna", # optional, safe retries
)
job_id = res["jobId"]
job = cs.get_job(job_id)
while job["status"] in ("queued", "processing"):
time.sleep(1.5)
job = cs.get_job(job_id)
if job["status"] == "failed":
raise RuntimeError(job.get("error") or "ingest failed")
results = cs.get_job_results(job_id, status="yellow")
for row in results["data"]:
print(row["memberRefId"], row["status"], row["flags"], row["commissionAmount"])If you upload an earlier month after a later one, the later period's scoring becomes stale. list_files() flags this with rescoreSuggested; refresh it without re-uploading:
files = cs.list_files(carrier_id="car_123")
for f in files["data"]:
if f.get("rescoreSuggested"):
cs.rescore_file(f["id"])Uploading over a carrier+period that already has a file fails with 409 (period_exists). To apply a corrected file, pass replace=True: the existing data is retracted and the corrected file re-ingested atomically. The following month is re-scored automatically.
res = cs.upload_file(
corrected_file,
carrier_id="car_123",
period_year=2026,
period_month=4,
replace=True, # omit -> 409 period_exists if the period already exists
)
# res["mode"] == "replace"
# Or remove a period entirely (no re-upload), re-scoring the next month:
cs.retract_file(file_id)status |
Meaning |
|---|---|
🟢 green |
Present and unchanged vs. the prior period. |
🟡 yellow |
Present but something tracked changed (see flags). |
🔴 red |
Present in the prior period, absent now (dropped). |
flag |
Meaning |
|---|---|
NEW |
First time this member is seen. |
COMMISSION_CHANGED |
Commission amount differs from the prior period. |
DATA_CHANGED |
A tracked non-commission field changed. |
DROPPED |
Was present before, missing now. |
REAPPEARED |
Returned after being absent. |
REAPPEARED_WITH_DELTA |
Returned and came back with a different commission. |
CHARGEBACK |
A negative-commission (clawback) record this period. |
from commissionsight import Status, Flag, ResultRow# Files & jobs
cs.list_files(carrier_id=carrier_id, limit=50)
cs.list_jobs(status="completed")
cs.get_job_results(job_id, status="red", limit=100, offset=0)
cs.get_job_deltas(job_id, change_type="COMMISSION_CHANGED")
cs.retry_job(job_id)
# Members & policies — status, timeline, and the full audit journey
cs.list_members(carrier_id=carrier_id, status="yellow")
cs.get_member_timeline(member_ref_id)
cs.get_member_journey(member_ref_id) # every period, source file, status + field changes
cs.get_policy_journey(policy_ref_id)
# Rejected rows from an ingest (exception file, as CSV text)
csv_text = cs.download_exceptions(job_id)
# Carriers & their mapping configs
cs.list_carriers(with_config=True)
cs.list_configs(carrier_id)A carrier group is a brand (e.g. "UHC") that fans out into several per-product member carriers (major medical, Medicare, ancillary). When you have a sample statement but aren't sure which member carrier it belongs to, resolve_carrier scores the brand's members by how well each one's config fits the file.
# Brands and their member carriers
groups = cs.list_carrier_groups()
print(groups["data"]) # [{"id", "name", "slug", "members": [...]}, ...]
# Resolve a brand + sample file to a concrete carrier.
# `file` can be a path, bytes, a file object, or a (filename, content) tuple.
res = cs.resolve_carrier("grp_uhc", "statements/sample.csv")
if res["ambiguous"]:
for cand in res["ranked"]: # too close to auto-pick — inspect candidates
print(cand["carrierId"], cand["productLine"], cand["confidence"], cand["reason"])
elif res["best"]:
print("matched", res["best"]["carrierId"], res["best"]["confidence"])cs.upsert_expected_rate(carrier_id=carrier_id, rate_type="percent_of_premium", rate_value=0.2)
rollup = cs.rollup("2026-05", carrier_id)
print(rollup["totals"]["commissionOwed"], rollup["totals"]["owedEvaluated"])cb = cs.list_chargebacks(carrier_id=carrier_id) # negative-commission events + original payoutcs.create_webhook(url="https://acme.com/hooks/cs", events=["job.completed"])cmp = cs.compare(from_period="2026-04", to_period="2026-05", carrier_id=carrier_id)
print(cmp["summary"]) # {"green", "yellow", "red", "new", "reappeared", "total"}cs.rollup("2026-05", carrier_id) # period totals by status + by carrier
cs.attrition("2026-05", carrier_id) # attrition rate for a period
cs.attrition_series(months=12) # attrition trend
cs.data_quality("2026-05") # statement-quality signals (ok/watch/alert)Roll up a whole range of periods into range-wide totals plus byPeriod and byCarrier breakdowns. Omit from_period/to_period for all history.
report = cs.cumulative(from_period="2026-01", to_period="2026-05", carrier_id=carrier_id)
print(report["totals"]["commissionOwed"], report["totals"]["memberMonths"])
for period in report["byPeriod"]:
print(period["period"], period["commissionAtRisk"])For multi-workspace accounts, list workspaces or create a new one. Pass the returned id as workspace_id to uploads and reports.
ws = cs.list_workspaces() # {"enabled": bool, "workspaces": [...]}
east = cs.create_workspace("East") # {"id", "name", "isDefault"}
cs.rollup("2026-05", workspace_id=east["id"])Admin endpoints (require an admin-role session) live under cs.admin:
cs.admin.list_accounts(status="pending")
cs.admin.account_overview(account_id)
cs.admin.metrics()
cs.admin.revenue()List endpoints return a data list plus an optional pagination object:
{
"data": [...],
"pagination": {"limit": 50, "offset": 0, "nextCursor": None, "hasMore": False},
}Offset-based endpoints accept limit / offset; cursor-based ones (e.g. list_files) accept limit / cursor and return nextCursor.
Any non-2xx response raises an ApiError, carrying the HTTP status and the parsed RFC 9457 problem+json body when present.
from commissionsight import ApiError
try:
cs.get_job("does-not-exist")
except ApiError as err:
print(err.status) # e.g. 404
print(str(err)) # problem `title`
print(err.body) # full problem+json payloadInject your own transport to use requests/httpx, add retries, or mock in tests. It takes (method, url, headers, body) and returns (status_code, response_text):
import requests
def transport(method, url, headers, body):
r = requests.request(method, url, headers=headers, data=body)
return r.status_code, r.text
cs = CommissionSightClient("https://api.commissionsight.com/v1", token="...", transport=transport)- Website: https://commissionsight.com
- API docs: https://docs.commissionsight.com
- TypeScript SDK: https://github.com/commissionsight/sdk
- Issues: https://github.com/commissionsight/python-sdk/issues
MIT © CommissionSight