"""Analyze OpenAlex publication data and print summary report.""" import numpy as np import pandas as pd from config import DATA_DIR, LAST_FULL_YEAR, COUNTRY_NAMES def pct_change(old, new): if old == 0: return 0.0 return (new - old) / old * 100 def fmt(n): """Format number with commas.""" return f"{n:,.0f}" def section(title): print() print("=" * 70) print(title) print("=" * 70) def validation(df): """Reproduce Hanson et al. 2024 result using OpenAlex.""" section("VALIDATION (Hanson et al. 2024, Scopus+WoS, 2016→2022)") full = df[(df["year"] >= 2016) & (df["year"] <= 2022)].sort_values("year").copy() c16, c22 = full.iloc[0]["count"], full.iloc[-1]["count"] cumul = pct_change(c16, c22) full["yoy"] = full["count"].pct_change() * 100 mean_yoy = full["yoy"].dropna().mean() # Exponential fit (log-linear OLS) for apples-to-apples comparison with Hanson's ~5.6% x = np.array(full["year"] - full["year"].min(), dtype=float) y = np.array(full["count"], dtype=float) coeffs = np.polyfit(x, np.log(y), 1) exp_rate = coeffs[0] * 100 print(f" Cumulative growth: {cumul:+.1f}% | Mean YoY: {mean_yoy:.1f}% | Exponential fit: {exp_rate:.1f}%/yr (OpenAlex, DOI-indexed)") def update(df): """Show what happened after Hanson's data ended (2022→2025).""" section(f"UPDATE (2022→{LAST_FULL_YEAR})") full = df[(df["year"] >= 2022) & (df["year"] <= LAST_FULL_YEAR)].sort_values("year").copy() c22, c_end = full.iloc[0]["count"], full.iloc[-1]["count"] cumul = pct_change(c22, c_end) full["yoy"] = full["count"].pct_change() * 100 mean_yoy = full["yoy"].dropna().mean() print(f" 2022: {fmt(c22)} articles") print(f" {LAST_FULL_YEAR}: {fmt(c_end)} articles") print(f" Cumulative growth: {cumul:+.1f}% | Mean YoY: {mean_yoy:.1f}%") # Full-period cumulative growth c2000 = df[df["year"] == 2000]["count"].iloc[0] c_latest = df[df["year"] == LAST_FULL_YEAR]["count"].iloc[0] full_cumul = pct_change(c2000, c_latest) print(f"\n Full period (2000→{LAST_FULL_YEAR}): {fmt(c2000)} → {fmt(c_latest)} ({full_cumul:+.1f}% cumulative, {c_latest/c2000:.1f}× increase)") def yoy_growth(df): """Print year-over-year growth rates.""" section("YEAR-OVER-YEAR GROWTH RATES") full = df[df["year"] <= LAST_FULL_YEAR].sort_values("year").copy() for i in range(1, len(full)): y = full.iloc[i]["year"] prev = full.iloc[i - 1]["count"] curr = full.iloc[i]["count"] yoy = pct_change(prev, curr) if y >= 2015: print(f" {int(y)}: {yoy:+5.1f}%") def field_analysis(df_domain): """Print domain and field-level analysis.""" section("FIELD ANALYSIS (2015→2025)") full = df_domain[df_domain["year"] <= LAST_FULL_YEAR].copy() y1, y2 = 2015, LAST_FULL_YEAR d1 = full[full["year"] == y1].groupby("domain_name")["count"].sum() d2 = full[full["year"] == y2].groupby("domain_name")["count"].sum() combined = pd.DataFrame({"count_start": d1, "count_end": d2}).dropna() combined["growth_pct"] = (combined["count_end"] - combined["count_start"]) / combined["count_start"] * 100 combined["added"] = combined["count_end"] - combined["count_start"] combined["share_start"] = combined["count_start"] / combined["count_start"].sum() * 100 combined["share_end"] = combined["count_end"] / combined["count_end"].sum() * 100 combined = combined.sort_values("count_end", ascending=False) header = f"{'Domain':<25} | {'Count '+str(y1):>12} | {'Count '+str(y2):>12} | {'Growth':>8} | {'Share '+str(y1):>9} | {'Share '+str(y2):>9}" print(header) print("-" * len(header)) for name, row in combined.iterrows(): print( f"{name:<25} | {fmt(row['count_start']):>12} | {fmt(row['count_end']):>12} | " f"{row['growth_pct']:>+7.1f}% | {row['share_start']:>8.1f}% | {row['share_end']:>8.1f}%" ) fastest = combined["growth_pct"].idxmax() most_added = combined["added"].idxmax() print(f"\n Fastest growing domain: {fastest} ({combined.loc[fastest, 'growth_pct']:+.1f}%)") print(f" Most papers added: {most_added} (+{fmt(combined.loc[most_added, 'added'])})") # Field-level breakdown (top 10 fastest growing) fields = pd.read_csv(DATA_DIR / "yearly_by_field.csv") fields_full = fields[fields["year"] <= LAST_FULL_YEAR].copy() f1 = fields_full[fields_full["year"] == y1].set_index("field_name")["count"] f2 = fields_full[fields_full["year"] == y2].set_index("field_name")["count"] fgrowth = pd.DataFrame({"count_start": f1, "count_end": f2}).dropna() fgrowth["growth_pct"] = (fgrowth["count_end"] - fgrowth["count_start"]) / fgrowth["count_start"] * 100 fgrowth = fgrowth.sort_values("growth_pct", ascending=False).head(10) print(f"\nFastest growing fields ({y1}→{y2}):") for i, (name, row) in enumerate(fgrowth.iterrows(), 1): print(f" {i:>2}. {name:<45} {fmt(row['count_start']):>10} → {fmt(row['count_end']):>10} ({row['growth_pct']:+.1f}%)") def country_analysis(df_country, df_totals): """Print country-level analysis.""" section("COUNTRY ANALYSIS (2015→2025)") print("Note: Shares use total articles as denominator (whole counting).") print("A co-authored paper counts once per country, so shares sum to >100%.") full = df_country[df_country["year"] <= LAST_FULL_YEAR].copy() global_total = df_totals[df_totals["year"] == LAST_FULL_YEAR]["count"].iloc[0] # Top 15 latest = full[full["year"] == LAST_FULL_YEAR].sort_values("count", ascending=False).head(15) print(f"\nTop 15 countries by {LAST_FULL_YEAR} output:") for i, (_, row) in enumerate(latest.iterrows(), 1): cc = row["country_code"] name = COUNTRY_NAMES.get(cc, cc) share = row["count"] / global_total * 100 print(f" {i:>2}. {cc} ({name:<20}): {fmt(row['count']):>12} ({share:.1f}%)") # China vs US print("\nChina vs United States:") for year in range(2000, LAST_FULL_YEAR + 1): yr_data = full[full["year"] == year] cn = yr_data[yr_data["country_code"] == "CN"]["count"].sum() us = yr_data[yr_data["country_code"] == "US"]["count"].sum() if cn > us and year >= 2010: cn_prev = full[(full["year"] == year - 1) & (full["country_code"] == "CN")]["count"].sum() us_prev = full[(full["year"] == year - 1) & (full["country_code"] == "US")]["count"].sum() if cn_prev <= us_prev: print(f" China surpassed the US in {year}") break cn_latest = full[(full["year"] == LAST_FULL_YEAR) & (full["country_code"] == "CN")]["count"].sum() us_latest = full[(full["year"] == LAST_FULL_YEAR) & (full["country_code"] == "US")]["count"].sum() print(f" China {LAST_FULL_YEAR}: {fmt(cn_latest)} ({cn_latest/global_total*100:.1f}% of articles)") print(f" US {LAST_FULL_YEAR}: {fmt(us_latest)} ({us_latest/global_total*100:.1f}% of articles)") if cn_latest > 0 and us_latest > 0: print(f" China/US ratio: {cn_latest/us_latest:.2f}x") # Fastest growing countries print(f"\nFastest growing countries (2015→{LAST_FULL_YEAR}, min 10k papers in 2015):") c2015 = full[full["year"] == 2015].set_index("country_code")["count"] clatest = full[full["year"] == LAST_FULL_YEAR].set_index("country_code")["count"] growth = pd.DataFrame({"c2015": c2015, "clatest": clatest}).dropna() growth = growth[growth["c2015"] >= 10000] growth["growth_pct"] = (growth["clatest"] - growth["c2015"]) / growth["c2015"] * 100 growth = growth.sort_values("growth_pct", ascending=False).head(10) for i, (cc, row) in enumerate(growth.iterrows(), 1): name = COUNTRY_NAMES.get(cc, cc) print(f" {i:>2}. {cc} ({name}): {row['growth_pct']:+.1f}%") def main(): totals = pd.read_csv(DATA_DIR / "yearly_totals.csv") domains = pd.read_csv(DATA_DIR / "yearly_by_domain.csv") countries = pd.read_csv(DATA_DIR / "yearly_by_country.csv") validation(totals) update(totals) yoy_growth(totals) field_analysis(domains) country_analysis(countries, totals) if __name__ == "__main__": main()