historical-network

Use this Skill for historical network analysis: correspondence networks, prosopographic data, temporal community detection, betweenness over time, and Gephi GEXF export.

仓库: xjtulyc/awesome-rosetta-skills
最近来源活动: 2026年3月18日 08:22
检测到的 SKILL.md 语言: 英语
星标: 34
分支: 6

安装方式

默认使用会先检查来源的 Prompt；你也可以切换为直接命令，或下载本地副本。

检查来源文件

决定是否安装前，请先阅读 SKILL.md，以及 SkillsMP 当前展示的配套文件。

返回搜索返回安装

name	historical-network
description	Use this Skill for historical network analysis: correspondence networks, prosopographic data, temporal community detection, betweenness over time, and Gephi GEXF export.
tags	["history","network-analysis","prosopography","temporal-networks","Gephi","digital-humanities"]
version	1.0.0
authors	[{"name":"awesome-rosetta-skills contributors","github":"@xjtulyc"}]
license	MIT
platforms	["claude-code","codex","gemini-cli","cursor"]
dependencies	{"python":["networkx>=3.1","pandas>=1.5","numpy>=1.23","matplotlib>=3.6","python-louvain>=0.16"]}
last_updated	2026-03-18
status	stable

Historical Network Analysis: Prosopography and Correspondence Networks

TL;DR — Build correspondence networks from archival letter datasets, attach prosopographic node attributes, compute temporal betweenness centrality by decade, detect communities per period with Louvain, measure community persistence via Jaccard, project bipartite person-event networks, and export to Gephi GEXF for visualization.

When to Use

Use this Skill when you need to:

Analyze the structure of historical correspondence networks (epistolary, diplomatic, merchant)
Study the rise and fall of influential intermediaries over decades or centuries
Detect scholarly, religious, or political communities across time periods
Visualize network evolution with node/edge attributes in Gephi
Project person × event membership matrices into person-person co-attendance networks

Do not use this Skill for:

Online social media graph analysis at millions-of-nodes scale (use GraphX or Snap.py)
Phylogenetic tree reconstruction (use dedicated bioinformatics tools)
Road or transport network optimization (use OR-Tools or igraph)

Background

Historical networks differ from modern social networks in two key ways:

Incompleteness: Only a fraction of historical correspondence survives. Network metrics must be interpreted with an eye on archival gaps.
Temporal dynamics: Polities, alliances, and scholarly circles shift over decades. Static network metrics hide this change; time-windowed analysis is essential.

Concept	Explanation
Prosopography	Systematic study of individuals in historical records; provides node attributes
Betweenness centrality	Fraction of shortest paths passing through a node; high = broker/gatekeeper
Louvain community detection	Modularity-maximizing algorithm; O(n log n)
Jaccard similarity
GEXF	Graph Exchange XML Format; native Gephi format supporting dynamic attributes
Bipartite projection	Person × event bipartite → person-person weighted unipartite graph

Environment Setup

# Create Python environment
conda create -n hist-net python=3.11 -y
conda activate hist-net


pip install    \
     python-louvain


python -c 
python -c

import networkx as nx
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from collections import defaultdict


def build_correspondence_network(
    letters_df: pd.DataFrame,
    sender_col: str = "sender_id",
    recipient_col: str = "recipient_id",
    date_col: str = "year",
    weight_threshold: int = 1,
) -> nx.MultiDiGraph:
    """
    Construct a directed multigraph from a historical correspondence dataset.

    Each letter becomes an edge with year and doc_id attributes.
    Parallel edges are preserved (one edge per letter).

    Args:
        letters_df:       DataFrame of letters with sender, recipient, year, doc_id columns.
        sender_col:       Column name for sender identifier.
        recipient_col:    Column name for recipient identifier.
        date_col:         Column name for the year integer.
        weight_threshold: Minimum number of letters to include an edge in the graph.

    Returns:
        Directed multigraph with letter-level edges.
    """
    G = nx.MultiDiGraph()

    # Add all unique persons as nodes
    all_persons = pd.concat([
        letters_df[sender_col],
        letters_df[recipient_col],
    ]).unique()
    G.add_nodes_from(all_persons)

    # Add edges
    for _, row in letters_df.iterrows():
        G.add_edge(
            row[sender_col],
            row[recipient_col],
            year=int(row[date_col]),
            doc_id=str(row.get("doc_id", "")),
        )

    # Optionally collapse to weighted DiGraph
    if weight_threshold > 1:
        simple = nx.DiGraph()
        edge_weights = defaultdict(int)
        for u, v, data in G.edges(data=True):
            edge_weights[(u, v)] += 1
        for (u, v), w in edge_weights.items():
            if w >= weight_threshold:
                simple.add_edge(u, v, weight=w)
        return simple

    return G


def add_prosopographic_attributes(
    G: nx.Graph,
    prosop_df: pd.DataFrame,
    id_col: str = "person_id",
) -> nx.Graph:
    """
    Add prosopographic metadata as node attributes to a network.

    Args:
        G:          NetworkX graph with person IDs as nodes.
        prosop_df:  DataFrame with columns: person_id, name, title, birth_year,
                    death_year, gender, institution.
        id_col:     Column name for the person identifier.

    Returns:
        Graph with updated node attributes (modifies in place).
    """
    attr_cols = [c for c in prosop_df.columns if c != id_col]
    for _, row in prosop_df.iterrows():
        pid = row[id_col]
        if pid in G.nodes:
            for col in attr_cols:
                G.nodes[pid][col] = row[col]
    return G


def temporal_betweenness(
    letters_df: pd.DataFrame,
    decade_size: int = 10,
    sender_col: str = "sender_id",
    recipient_col: str = "recipient_id",
    date_col: str = "year",
) -> pd.DataFrame:
    """
    Compute betweenness centrality per decade window.

    For each decade, build an undirected weighted graph from letters
    in that window and compute normalized betweenness centrality.

    Args:
        letters_df:  DataFrame of letters.
        decade_size: Width of time window in years.
        sender_col:  Column for sender IDs.
        recipient_col: Column for recipient IDs.
        date_col:    Column for year.

    Returns:
        DataFrame with columns: person_id, decade_start, betweenness.
    """
    min_year = int(letters_df[date_col].min())
    max_year = int(letters_df[date_col].max())
    decades = range(min_year, max_year, decade_size)

    records = []
    for decade_start in decades:
        decade_end = decade_start + decade_size
        window = letters_df[
            (letters_df[date_col] >= decade_start) &
            (letters_df[date_col] < decade_end)
        ]
        if window.empty:
            continue

        # Build weighted undirected graph for this decade
        G_dec = nx.Graph()
        for _, row in window.iterrows():
            u, v = row[sender_col], row[recipient_col]
            if G_dec.has_edge(u, v):
                G_dec[u][v]["weight"] += 1
            else:
                G_dec.add_edge(u, v, weight=1)

        if G_dec.number_of_nodes() < 3:
            continue

        bc = nx.betweenness_centrality(G_dec, weight="weight", normalized=True)
        for person_id, score in bc.items():
            records.append({
                "person_id": person_id,
                "decade_start": decade_start,
                "betweenness": round(score, 6),
            })

    return pd.DataFrame(records)


def plot_betweenness_over_time(
    betweenness_df: pd.DataFrame,
    top_n: int = 5,
    prosop_df: pd.DataFrame = None,
    output_path: str = None,
) -> None:
    """
    Line plot of top-N persons by peak betweenness centrality over decades.

    Args:
        betweenness_df: DataFrame from temporal_betweenness().
        top_n:          Number of top persons to highlight.
        prosop_df:      Optional prosopographic DataFrame; if given, use 'name' column.
        output_path:    If given, save figure here.
    """
    peak_bc = (
        betweenness_df.groupby("person_id")["betweenness"]
        .max()
        .sort_values(ascending=False)
        .head(top_n)
    )
    top_persons = peak_bc.index.tolist()

    fig, ax = plt.subplots(figsize=(11, 5))
    for pid in top_persons:
        sub = betweenness_df[betweenness_df["person_id"] == pid].sort_values("decade_start")
        label = pid
        if prosop_df is not None and "name" in prosop_df.columns:
            name_row = prosop_df[prosop_df.iloc[:, 0] == pid]
            if not name_row.empty:
                label = name_row.iloc[0]["name"]
        ax.plot(sub["decade_start"], sub["betweenness"], marker="o", label=label)

    ax.set_xlabel("Decade start")
    ax.set_ylabel("Betweenness centrality (normalized)")
    ax.set_title(f"Top {top_n} Brokers by Betweenness Centrality Over Time")
    ax.legend(fontsize=8, loc="upper left")
    ax.grid(True, alpha=0.3)
    fig.tight_layout()

    if output_path:
        fig.savefig(output_path, dpi=150)
        print(f"Betweenness plot saved to {output_path}")
    plt.show()

import community as community_louvain


def detect_communities_per_period(
    letters_df: pd.DataFrame,
    decade_size: int = 10,
    sender_col: str = "sender_id",
    recipient_col: str = "recipient_id",
    date_col: str = "year",
    resolution: float = 1.0,
) -> dict:
    """
    Detect Louvain communities for each decade window.

    Args:
        letters_df:  DataFrame of letters.
        decade_size: Width of time window in years.
        sender_col:  Column for sender IDs.
        recipient_col: Column for recipient IDs.
        date_col:    Column for year integer.
        resolution:  Louvain resolution parameter (>1 = more smaller communities).

    Returns:
        Dict mapping decade_start (int) → community_partition (dict: node → community_id).
    """
    min_year = int(letters_df[date_col].min())
    max_year = int(letters_df[date_col].max())

    period_communities = {}
    for decade_start in range(min_year, max_year, decade_size):
        decade_end = decade_start + decade_size
        window = letters_df[
            (letters_df[date_col] >= decade_start) &
            (letters_df[date_col] < decade_end)
        ]
        if window.empty:
            continue

        G_dec = nx.Graph()
        for _, row in window.iterrows():
            u, v = row[sender_col], row[recipient_col]
            if G_dec.has_edge(u, v):
                G_dec[u][v]["weight"] += 1
            else:
                G_dec.add_edge(u, v, weight=1)

        if G_dec.number_of_nodes() < 4:
            continue

        partition = community_louvain.best_partition(
            G_dec, weight="weight", resolution=resolution, random_state=42
        )
        period_communities[decade_start] = partition

    return period_communities


def compute_community_jaccard_persistence(
    period_communities: dict,
) -> pd.DataFrame:
    """
    Measure the persistence of communities across consecutive periods using Jaccard similarity.

    For each pair of consecutive periods, compute the maximum Jaccard similarity
    between every community in period T and every community in period T+1.
    High Jaccard = stable community; low Jaccard = community dissolved.

    Args:
        period_communities: Dict from detect_communities_per_period().

    Returns:
        DataFrame with columns: period_a, period_b, community_a, community_b,
        jaccard, matched_count.
    """
    periods = sorted(period_communities.keys())
    records = []

    for i in range(len(periods) - 1):
        pa = periods[i]
        pb = periods[i + 1]

        partition_a = period_communities[pa]
        partition_b = period_communities[pb]

        # Group nodes by community
        def group_by_comm(part):
            groups = defaultdict(set)
            for node, comm in part.items():
                groups[comm].add(node)
            return groups

        groups_a = group_by_comm(partition_a)
        groups_b = group_by_comm(partition_b)

        for comm_a, nodes_a in groups_a.items():
            best_jacc = 0.0
            best_comm_b = None
            for comm_b, nodes_b in groups_b.items():
                intersection = len(nodes_a & nodes_b)
                union = len(nodes_a | nodes_b)
                jacc = intersection / union if union > 0 else 0.0
                if jacc > best_jacc:
                    best_jacc = jacc
                    best_comm_b = comm_b

            records.append({
                "period_a": pa,
                "period_b": pb,
                "community_a": comm_a,
                "community_b": best_comm_b,
                "jaccard": round(best_jacc, 4),
                "size_a": len(nodes_a),
            })

    return pd.DataFrame(records)

def build_annotated_network_for_export(
    letters_df: pd.DataFrame,
    prosop_df: pd.DataFrame = None,
    period_communities: dict = None,
    sender_col: str = "sender_id",
    recipient_col: str = "recipient_id",
    date_col: str = "year",
) -> nx.Graph:
    """
    Build a weighted undirected graph with full node/edge attributes for Gephi export.

    Node attributes: name, gender, institution, community_id (from last period).
    Edge attributes: weight (letter count), first_year, last_year.

    Args:
        letters_df:        DataFrame of letters.
        prosop_df:         Optional prosopographic DataFrame.
        period_communities: Optional dict from detect_communities_per_period().
        sender_col:        Column for sender IDs.
        recipient_col:     Column for recipient IDs.
        date_col:          Column for year.

    Returns:
        Annotated undirected NetworkX Graph ready for nx.write_gexf().
    """
    G = nx.Graph()

    # Build weighted edges
    edge_data = defaultdict(lambda: {"weight": 0, "years": []})
    for _, row in letters_df.iterrows():
        u, v = str(row[sender_col]), str(row[recipient_col])
        key = tuple(sorted([u, v]))
        edge_data[key]["weight"] += 1
        edge_data[key]["years"].append(int(row[date_col]))

    all_nodes = set()
    for (u, v), data in edge_data.items():
        all_nodes.update([u, v])
        G.add_edge(
            u, v,
            weight=data["weight"],
            first_year=min(data["years"]),
            last_year=max(data["years"]),
        )

    # Add prosopographic node attributes
    if prosop_df is not None:
        id_col = prosop_df.columns[0]
        for _, row in prosop_df.iterrows():
            pid = str(row[id_col])
            if pid in G.nodes:
                for col in prosop_df.columns:
                    if col != id_col:
                        val = row[col]
                        if pd.notna(val):
                            G.nodes[pid][col] = str(val)

    # Add community labels from the last period
    if period_communities:
        last_period = max(period_communities.keys())
        partition = period_communities[last_period]
        for node, comm_id in partition.items():
            node_str = str(node)
            if node_str in G.nodes:
                G.nodes[node_str]["community"] = int(comm_id)

    return G


def export_gexf(
    G: nx.Graph,
    output_path: str,
) -> None:
    """
    Export a NetworkX graph to GEXF format for Gephi visualization.

    GEXF supports node/edge attributes and can encode dynamic networks
    with time intervals. Open the output file directly in Gephi.

    Args:
        G:           Annotated NetworkX graph.
        output_path: Absolute path to write the .gexf file.
    """
    nx.write_gexf(G, output_path)
    print(f"GEXF file written to {output_path}")
    print(f"  Nodes: {G.number_of_nodes()}")
    print(f"  Edges: {G.number_of_edges()}")
    print(f"  Open in Gephi: File → Open → {output_path}")

def build_bipartite_person_event(
    attendance_df: pd.DataFrame,
    person_col: str = "person_id",
    event_col: str = "event_id",
) -> tuple[nx.Graph, nx.Graph]:
    """
    Build a bipartite person-event graph and project to person-person co-attendance.

    Edge weight in the projected graph = number of events co-attended.

    Args:
        attendance_df: DataFrame with person_id and event_id columns.
        person_col:    Column for person identifiers.
        event_col:     Column for event identifiers.

    Returns:
        Tuple of (bipartite_graph, person_person_projection).
    """
    from networkx.algorithms import bipartite

    B = nx.Graph()

    persons = attendance_df[person_col].unique()
    events = attendance_df[event_col].unique()

    B.add_nodes_from(persons, bipartite=0)
    B.add_nodes_from(events, bipartite=1)

    for _, row in attendance_df.iterrows():
        B.add_edge(row[person_col], row[event_col])

    # Project onto persons (bipartite=0 set)
    person_nodes = {n for n, d in B.nodes(data=True) if d.get("bipartite") == 0}
    projected = bipartite.weighted_projected_graph(B, person_nodes)
    return B, projected

Problem	Cause	Fix
`ModuleNotFoundError: community`	python-louvain not installed	`pip install python-louvain`
Betweenness very slow on large graph	O(n*m) algorithm	Use `nx.betweenness_centrality(G, k=100)` for approximate BC
GEXF file won't open in Gephi	Non-string node IDs	Cast all node IDs to `str()` before `write_gexf()`
Louvain non-deterministic results	Random seed not set	Pass `random_state=42` to `best_partition()`
`NetworkXError: Graph has no edges` on thin decade	Very few letters in that window	Increase `decade_size` or skip windows with fewer than 5 edges
Community Jaccard all near 0	Network turnover too high	Check for person ID inconsistencies in source data

import pandas as pd
import numpy as np

# Simulate a historical correspondence dataset (1600–1700)
np.random.seed(42)
n_letters = 500
persons = [f"P{i:03d}" for i in range(1, 51)]
years = np.random.randint(1600, 1700, n_letters)
senders = np.random.choice(persons, n_letters)
recipients = np.random.choice(persons, n_letters)

letters_df = pd.DataFrame({
    "sender_id": senders,
    "recipient_id": recipients,
    "year": years,
    "doc_id": [f"DOC_{i:04d}" for i in range(n_letters)],
})
# Remove self-loops
letters_df = letters_df[letters_df["sender_id"] != letters_df["recipient_id"]].reset_index(drop=True)

# Compute temporal betweenness
bc_df = temporal_betweenness(letters_df, decade_size=10)
print("Top brokers by peak betweenness:")
peak = bc_df.groupby("person_id")["betweenness"].max().sort_values(ascending=False).head(5)
print(peak)

# Plot betweenness over time
plot_betweenness_over_time(bc_df, top_n=5, output_path="/data/output/betweenness_timeline.png")

# Detect communities per decade
communities = detect_communities_per_period(letters_df, decade_size=20)
print(f"Periods analyzed: {sorted(communities.keys())}")

# Community persistence
jaccard_df = compute_community_jaccard_persistence(communities)
stable = jaccard_df[jaccard_df["jaccard"] > 0.4]
print(f"\nStable community links (Jaccard > 0.4): {len(stable)}")
print(jaccard_df.sort_values("jaccard", ascending=False).head(10).to_string(index=False))

# Build full annotated graph and export for Gephi
G_full = build_annotated_network_for_export(
    letters_df, period_communities=communities
)
export_gexf(G_full, "/data/output/republic_of_letters.gexf")

print("\nOpen /data/output/republic_of_letters.gexf in Gephi.")
print("Use 'ForceAtlas 2' layout and color nodes by 'community' attribute.")