File size: 5,880 Bytes
1025a87 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | from dataclasses import replace
import networkx as nx
import numpy as np
from socialdynamics.experiments import aggregate, beta_sweep, run_ensemble, summarise_sweep
from socialdynamics.presets import PRESETS
from socialdynamics.simulation import OpinionSimulation, SimulationConfig
def _preset_config(name: str, **overrides) -> SimulationConfig:
p = dict(PRESETS[name])
p["misinformation_direction"] = 1.0 if p["misinformation_direction"].startswith("Positive") else -1.0
p.update(overrides)
return SimulationConfig(**p)
def _reference_step_v11(sim: OpinionSimulation, old: np.ndarray) -> np.ndarray:
"""The v1.1 per-agent loop, kept verbatim as a regression oracle."""
cfg = sim.config
new = old.copy()
beta = 6.0 * cfg.confirmation_bias
for agent in range(cfg.n_agents):
if sim.stubborn_mask[agent]:
continue
neighbors = list(sim.graph.neighbors(agent))
if neighbors:
w = np.exp(-beta * np.abs(old[agent] - old[neighbors]))
total = float(np.sum(w))
target = float(np.sum(w * old[neighbors]) / total) if total > 1e-12 else float(old[agent])
else:
target = float(old[agent])
delta = cfg.social_influence * (target - old[agent])
if sim.misinformation_mask[agent] and cfg.misinformation_strength > 0:
delta += 0.22 * cfg.misinformation_strength * (cfg.misinformation_direction - old[agent])
if cfg.noise > 0:
delta += float(sim._ref_rng.normal(0.0, cfg.noise))
new[agent] = float(np.clip(old[agent] + delta, -1.0, 1.0))
return new
def test_vectorised_update_matches_v11_loop():
for kw in ({}, {"network": "Scale Free", "initial_opinion": "Polarized"}, {"stubborn_fraction": 0.2}):
cfg = SimulationConfig(steps=40, seed=17, **kw)
fast = OpinionSimulation(cfg)
ref = OpinionSimulation(cfg)
ref._ref_rng = ref.rng
x = ref.opinions.copy()
for _ in range(cfg.steps):
fast.step()
x = _reference_step_v11(ref, x)
assert np.max(np.abs(fast.opinions - x)) < 1e-12
def test_simulation_is_seed_deterministic():
for rate in (0.0, 0.2):
cfg = SimulationConfig(n_agents=100, steps=20, seed=123, rewiring_rate=rate)
a, b = OpinionSimulation(cfg), OpinionSimulation(cfg)
a.run()
b.run()
assert np.array_equal(a.history_array, b.history_array)
assert set(map(frozenset, a.graph.edges())) == set(map(frozenset, b.graph.edges()))
def test_opinions_stay_bounded():
cfg = SimulationConfig(n_agents=100, steps=50, seed=3, misinformation_exposure=1.0,
misinformation_strength=1.0, social_influence=1.0, noise=0.2)
sim = OpinionSimulation(cfg)
sim.run()
assert np.all(sim.history_array >= -1.0)
assert np.all(sim.history_array <= 1.0)
def test_stubborn_agents_never_move():
cfg = SimulationConfig(n_agents=100, steps=30, seed=4, stubborn_fraction=0.20, stubborn_mode="Split extremes",
misinformation_exposure=0.50, misinformation_strength=0.8, rewiring_rate=0.3)
sim = OpinionSimulation(cfg)
initial = sim.opinions.copy()
mask = sim.stubborn_mask.copy()
sim.run()
assert np.array_equal(sim.opinions[mask], initial[mask])
def test_misinformation_can_shift_mean_positive():
base = dict(n_agents=100, network="Random", initial_opinion="Neutral", stubborn_fraction=0.0,
confirmation_bias=0.2, social_influence=0.4, noise=0.0, steps=30, seed=11)
control = OpinionSimulation(SimulationConfig(**base, misinformation_exposure=0.0, misinformation_strength=0.0))
treatment = OpinionSimulation(SimulationConfig(**base, misinformation_exposure=1.0, misinformation_strength=0.8))
control.run()
treatment.run()
assert treatment.opinions.mean() > control.opinions.mean() + 0.1
def test_rewiring_conserves_edges_and_creates_no_isolates():
sim = OpinionSimulation(_preset_config("Echo Chamber"))
edges0 = sim.graph.number_of_edges()
degree0 = dict(sim.graph.degree())
sim.run()
assert sim.rewired_ties > 0
assert sim.graph.number_of_edges() == edges0
assert not list(nx.isolates(sim.graph))
assert nx.number_of_selfloops(sim.graph) == 0
assert sum(dict(sim.graph.degree()).values()) == sum(degree0.values())
def test_rewiring_does_not_perturb_the_noise_stream():
cfg = SimulationConfig(steps=1, seed=8)
a = OpinionSimulation(cfg)
b = OpinionSimulation(replace(cfg, rewiring_rate=0.5))
a.step()
b.step()
assert np.array_equal(a.opinions, b.opinions) # first update precedes any rewiring
def test_filter_alone_does_not_segregate_but_rewiring_does():
seeds = [42, 43, 44]
filtered = run_ensemble(_preset_config("Filtered, Not Segregated"), seeds, n_permutations=100)
chamber = run_ensemble(_preset_config("Echo Chamber"), seeds, n_permutations=100)
for run in filtered:
assert run["selective_exposure"] > 0.9 # strong filter...
assert abs(run["assortativity"]) < 0.2 # ...but no clustering
assert run["assortativity_z"] < 3.0
for run in chamber:
assert run["assortativity"] > 0.9
assert run["assortativity_z"] > 10.0
# and the old index would have ranked this *below* the unsegregated case
assert run["selective_exposure"] < min(r["selective_exposure"] for r in filtered)
def test_sweep_and_aggregate_shapes():
base = SimulationConfig(steps=20)
rows = beta_sweep(base, (0.0, 1.0), (0.0, 0.1), seeds=[1, 2], n_permutations=20)
assert len(rows) == 8
summary = summarise_sweep(rows, ("assortativity", "selective_exposure"))
assert len(summary) == 4
assert all(r["n"] == 2 for r in summary)
agg = aggregate(rows, ("assortativity",))
assert agg["assortativity_ci95"] >= 0.0
|