Pipeline Documentation: Systemic Risk Channel Scoring

CALCULATION

• Computes min_delay = 1.0 / requests_per_second (e.g. 0.1s at 10 req/s)
• Initializes last_request_time = 0

CALCULATION

1. Read current time: current_time = time.time()
2. Compute elapsed: time_since_last = current_time - self.last_request_time
3. If time_since_last < self.min_delay: sleep for the difference
4. Update self.last_request_time = time.time()

CALCULATION

1. Default params to {} if None
2. If self.email is set, inject params['mailto'] = self.email
3. Build URL: urljoin(BASE_URL, endpoint)
4. For each attempt 0..max_retries-1:
   • Call self._rate_limit()
   • Execute requests.get(url, params=params, timeout=30)
   • HTTP 200 → return response.json()
   • HTTP 403 (rate limited) → sleep 2^attempt seconds, retry
   • HTTP 5xx (server error) → sleep 2^attempt seconds, retry
   • Other 4xx → raise immediately (no retry)
   • Timeout / ConnectionError → sleep 2^attempt seconds, retry (or raise on last attempt)
5. If all retries exhausted → raise Exception

CALCULATION

1. Build params dict: {'per-page': min(per_page, 200), 'page': page}
2. If search: add params['search'] = search
3. If filter_params: join as comma-separated key:value string into params['filter']
4. If sort: add params['sort'] = sort
5. If select: join as comma-separated string into params['select']
6. Delegate to self._make_request('/works', params)

Example constructed URL:
/works?search=systemic+risk+contagion&filter=publication_year:2009-2026,cited_by_count:>4&sort=cited_by_count:desc&per-page=200&page=1

CALCULATION

Constructs endpoint /{entity_type}/{entity_id} and delegates to _make_request().

CALCULATION

1. Iterate over ids in chunks of 50
2. For each chunk: join with | separator
3. Construct filter: {id_field}:{id1|id2|...|id50}
4. Call _make_request(/{entity_type}, params={'filter':..., 'per-page':50})
5. Accumulate all results arrays

CALCULATION

1. Copy params, set per-page=200, page=1
2. Loop:
   • Fetch page via _make_request()
   • Extend accumulator with results
   • If max_results reached → truncate and return
   • If len(all_results) ≥ meta.count → break
   • Increment page
   • If page × 200 > 10000 → break (OpenAlex hard limit)

CALCULATION

Case A: sample_size ≤ 10,000 (standard path)

1. Set params = {sample: sample_size, per-page: 200, seed: seed}
2. Paginate through ceil(sample_size / 200) pages
3. Deduplicate by work ID
4. Truncate to requested size

Case B: sample_size > 10,000 (multi-seed path)

1. Iterate over seed offsets: current_seed = seed + i
2. Each seed fetches up to 10,000 results with internal pagination (max 5 pages per seed)
3. Deduplicate across all seeds
4. Stop when target reached

CALCULATION

1. Build params: {group_by: group_field}
2. If filter_params: build comma-separated filter string
3. Call _make_request(/{entity_type}, params)
4. Extract response['group_by']

CALCULATION

1. Authors: Iterate authorships, collect each author.display_name
2. Concepts: Iterate concepts, collect each display_name
3. Abstract reconstruction:
   • For each (word, positions) in the inverted index: create tuples (position, word)
   • Sort by position ascending
   • Join words with spaces
4. Host venue: Navigate primary_location → source → display_name

CALCULATION

1. Initialize seen_ids = set(), channel_papers = []
2. For each query in channel_info['queries']:
   • If len(channel_papers) ≥ per_channel_limit: stop
   • Construct filter: {publication_year: "YYYY-YYYY", cited_by_count: ">N"}
   • Call client.search_works(search=query, filter_params=..., per_page=remaining, sort="cited_by_count:desc")
   • Extract and deduplicate results via extract_paper()
3. Pagination: If total_available > per_page and limit not reached:
   • Compute pages needed (capped at 4 additional pages)
   • Fetch pages 2 through pages_needed + 1
   • Continue deduplicating and accumulating

CALCULATION

1. Parse year range string into year_start, year_end
2. Load search_queries.json, extract channels dict (14 channels)
3. Initialize OpenAlexClient(email=...)
4. For each channel:
   • Call search_channel()
   • Write per-channel file: openalex_raw_{channel_id}.json
   • Merge into global dict keyed by paper ID
   • Track which channels found each paper in merged_papers[pid]['channels'] list
5. Convert merged dict to list, save as openalex_merged.json
6. Print summary: per-channel counts, total unique papers, year distribution histogram, low-result warnings

CALCULATION

1. Initialize channel_papers = defaultdict(list)
2. For each paper:
   • channels = paper.get("channels", [])
   • If channels exist: paper["primary_channel"] = channels[0]
   • paper["secondary_channels"] = channels[1:] (empty list if only one channel)
   • If no channels: paper["primary_channel"] = "unassigned"
   • For each channel in the paper's channel list: channel_papers[ch].append(paper)

Note: A paper appears in multiple channel_papers lists if it was found by multiple channel queries. This is intentional — it means the paper contributes to volume and impact scores for all channels that found it.

CALCULATION

1. Count papers per channel: counts[ch] = len(papers)
2. Find maximum: max_count = max(counts.values()) (fallback: 1 if empty)
3. Normalize each channel: lit_volume[ch] = counts[ch] / max_count

Properties: All values in [0, 1]. Exactly one channel scores 1.0 (the one with the most papers). Monotonically increasing with paper count.

CALCULATION

1. For each channel:
   • Sort papers by cited_by_count descending
   • Take the top 10 papers (or fewer if channel has <10)
   • Compute mean: sum(cited_by_count for top 10) / 10
   • Important: Always divides by 10, even if fewer than 10 papers exist. This penalizes channels with few papers.
2. Find maximum mean across all channels: max_mean
3. If max_mean == 0: set to 1 (prevent division by zero)
4. Normalize: cit_impact[ch] = channel_mean[ch] / max_mean

Edge cases:

• Empty channel (no papers) → mean_cites = 0
• Channel with 3 papers having 100, 50, 20 citations → mean = (100+50+20)/10 = 17.0 (not 56.7)
• All channels have zero citations → all scores = 0.0

CALCULATION

1. Compute median loss:
   • Parse all losses_usd strings that are valid floats
   • Sort numerically, compute median
   • Fallback: $100,000,000 if no numeric losses exist
2. For each crisis event:
   • Parse losses_usd. If non-numeric (e.g. "undetermined") → use median loss
   • Compute weight: log10(losses_usd)
   • If losses_usd ≤ 0: weight = 0
   • For each channel in channels_activated (or channels fallback):
     • If channel in all_channel_ids: crisis_counts[ch] += weight
3. Warn about channels in events that are not in all_channel_ids
4. Ensure all channels have entries (default 0 for channels with no crisis events)
5. Normalize: crisis_ev[ch] = crisis_counts[ch] / max(crisis_counts)

Edge cases:

• losses_usd = "undetermined" → uses median of all known losses
• losses_usd = "0" or negative → weight = 0
• No crisis events at all → all channels get 0.0
• max crisis count = 0 → treated as 1 (prevents division by zero)
• Unknown channel ID in event → skipped with printed warning

CALCULATION

1. Load data: Read all three JSON files. Crisis file is optional (missing = zero crisis scores + warning).
2. Assign channels: assign_channels(papers) → get reverse mapping
3. Build channel ID set: Union of search_queries keys and any IDs found in papers
4. Compute sub-scores:
   • compute_literature_volume(channel_papers)
   • compute_citation_impact(channel_papers)
   • compute_crisis_evidence(crisis_events, all_channel_ids)
5. Composite score (primary weights):

6. Rank: Sort channels by raw composite score descending, assign ranks 1..N, then round scores to 4 decimal places
7. Enrichment: Each ranking entry also includes paper_count, mean_top10_citations (rounded to 1 decimal), and crisis_event_count
8. Sensitivity analysis: Recompute composite scores under 4 weight schemes:

For each scheme, ranks are independently computed from the reweighted composite scores.