Go to Qiita Advent Calendar Top
LoginSignup
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@innovation1005

Python3ではじめるシステムトレード：リスク管理 ーエージェントシミュレーション

Last updated at Posted at 2026-01-01

image.png

1. 登場人物と3つの「社会の形」

このシミュレーションでは、600人の「人」が四角いエリアの中に住んでいます。彼らは3つの異なるパターンの社会を作ります。

  • ① 多様な社会（Diverse society）

  • みんながバラバラに、自分の好きな場所に住んでいる状態です。

  • ② 小グループに分かれた社会（Society with multiple groups）

  • いくつかの小さなコミュニティ（村や町のようなもの）に分かれて固まって住んでいる状態です。

  • ③ 完全に統合された社会（Fully unified society）

  • 全員が1か所にギュッと集まって、ひとつの大きな塊として住んでいる状態です。

2. ストーリーの流れ

シミュレーションは3つのステップで進みます。

  1. 平時（Normal times）： それぞれの社会の形で、安定して暮らしています。
  2. ショック発生（Shock）： 突然、画面に「安全地帯（円）」が現れます。この円の外にいる人は「危険な状態」になります。
  3. 復旧・救助（Recovery）： 安全地帯がじわじわと広がり、外にいる人たちが円の中に逃げ込もうと移動します。

3. このモデルが示している「教訓」

このシミュレーションの面白いところは、「ひとつの場所にまとまっていることが、必ずしも安全とは限らない」という点を見せていることです。

特に「③ 完全に統合された社会」に注目してください。

  • もし、全員が集まっている場所がたまたま「安全地帯」から外れてしまったら、その社会のほぼ全員（100%に近い人数）が一度に危機に陥ってしまいます。
  • 一方で、「① 多様な社会」や「② 小グループの社会」では、場所が分散しているため、一部の人は円の外に出てしまいますが、最初から円の中にいて無事な人も必ず一定数存在します。

4. まとめ

グラフの右下に出るメッセージ「Outcome depends strongly on where the group is located（結果は、グループがどこにいるかに強く依存する）」が、このモデルの核心です。

  • みんなで1か所に集まる（団結する）ことは、平時は効率が良いかもしれませんが、災害時には「全滅」のリスクを孕んでいます。
  • バラバラに住む（分散する）ことは、誰かは困るかもしれませんが、社会全体が一度にダメになるリスクを避ける「リスク分散」の役割を果たしています。

一言でいうと、「リスク管理における『卵をひとつのカゴに盛るな』という教訓を、社会の形で表現したモデル」と言えます。

import numpy as np
import matplotlib.pyplot as plt
from matplotlib.patches import Circle

# ============================================================
# 0) Global settings
# ============================================================
SEED = 7
rng = np.random.default_rng(SEED)

N = 600
T = 220

L_SIM  = 1.10   # reflecting wall (simulation)
L_VIEW = 1.35   # plotting range (bigger so circles are never clipped)

# storyboard times
t_form  = 80
t_shock = 120
rescue_len = 50
t_end = min(T - 1, t_shock + rescue_len)

# Safe zone (circle): appears at shock, expands during recovery
C = np.array([0.25, 0.05], float)
R0 = 0.55
R1 = 0.90

# Rescue dynamics (capacity-limited pull toward center during recovery)
RESCUE_CAP_PER_STEP = 60
RESCUE_PULL = 0.18

# Multiple groups
K_GROUPS = 6
MIN_SEP = 0.60

# Fully unified society (single cluster) placed "unluckily" to show location risk
HOME_ONE = np.array([-0.70, 0.05], float)

# ============================================================
# 1) Helper: reflecting boundary
# ============================================================
def reflect_1d(pos, vel, L):
    span = 2.0 * L
    x = pos + L
    n = np.floor(x / span).astype(np.int64)
    r = x - n * span
    odd = (n % 2) == 1
    pos2 = np.where(odd, span - r, r) - L
    vel2 = np.where(odd, -vel, vel)
    return pos2, vel2

def reflect_square(p, v, L):
    p[:, 0], v[:, 0] = reflect_1d(p[:, 0], v[:, 0], L)
    p[:, 1], v[:, 1] = reflect_1d(p[:, 1], v[:, 1], L)
    return p, v

def inside_circle(p, c, R):
    return np.linalg.norm(p - c[None, :], axis=1) <= R

# ============================================================
# 2) Same people across societies
# ============================================================
p0 = rng.uniform(-0.95, 0.95, size=(N, 2))
trait = rng.beta(2.0, 2.0, size=N)  # fixed diversity attribute (color only)

# ============================================================
# 3) Group hubs + fixed membership
# ============================================================
def sample_hubs(K, low=-0.85, high=0.85, min_sep=0.55, seed=0,
                max_tries=60000, relax=0.97, min_sep_floor=0.20):
    base_seed = int(seed)
    sep = float(min_sep)
    while sep >= min_sep_floor:
        r = np.random.default_rng(base_seed)
        hubs = []
        tries = 0
        while len(hubs) < K and tries < max_tries:
            x = r.uniform(low, high, size=2)
            if len(hubs) == 0:
                hubs.append(x)
            else:
                d = np.linalg.norm(np.asarray(hubs) - x[None, :], axis=1)
                if np.all(d >= sep):
                    hubs.append(x)
            tries += 1
        if len(hubs) == K:
            return np.asarray(hubs, float)
        sep *= relax
    raise RuntimeError("Could not place hubs. Reduce K or widen the area.")

hubs = sample_hubs(K_GROUPS, min_sep=MIN_SEP, seed=SEED + 11)

d2 = ((p0[:, None, :] - hubs[None, :, :])**2).sum(axis=2)
z = d2.argmin(axis=1)

# refine hubs to initial centroids (natural)
for k in range(K_GROUPS):
    m = (z == k)
    if m.any():
        hubs[k] = p0[m].mean(axis=0)

# ============================================================
# 4) Simulation
# ============================================================
def R_safe(t):
    if t < t_shock:
        return None
    if t >= t_end:
        return R1
    s = (t - t_shock) / max(1, (t_end - t_shock))
    return R0 + (R1 - R0) * s

def simulate(kind, seed=0):
    """
    kind: 'diverse' | 'groups' | 'one'
    """
    r = np.random.default_rng(seed)
    p = p0.copy()
    v = r.normal(0.0, 0.01, size=(N, 2))
    P = np.zeros((T, N, 2), float)

    # formation ramp
    ramp_start = 10
    ramp_len   = 60

    # parameters
    if kind == "diverse":
        noise = 0.003
        anchor_k = 0.090
        tau_group = 0.0
        tau_one   = 0.0
    elif kind == "groups":
        noise = 0.006
        anchor_k = 0.025
        tau_group = 0.055
        tau_one   = 0.0
    elif kind == "one":
        noise = 0.006
        anchor_k = 0.000
        tau_group = 0.0
        tau_one   = 0.085
    else:
        raise ValueError("kind must be 'diverse', 'groups', or 'one'")

    OUTSIDE_NOISE_MULT = 1.7
    OUTSIDE_DRAG = 0.12
    vmax = 0.22

    for t in range(T):
        rr = 0.0 if t < ramp_start else min(1.0, (t - ramp_start) / ramp_len)

        Rs = R_safe(t)
        safe = np.zeros(N, dtype=bool) if Rs is None else inside_circle(p, C, Rs)
        outside = ~safe if Rs is not None else np.zeros(N, dtype=bool)

        eps = r.normal(0.0, noise, size=(N, 2))
        if np.any(outside):
            eps[outside] *= OUTSIDE_NOISE_MULT

        anchor = anchor_k * (p0 - p)

        group_force = np.zeros_like(p)
        if kind == "groups":
            group_force = rr * tau_group * (hubs[z] - p)

        one_force = np.zeros_like(p)
        if kind == "one":
            one_force = rr * tau_one * (HOME_ONE[None, :] - p)

        pull = np.zeros_like(p)
        if (Rs is not None) and (t_shock <= t < t_end):
            idx = np.where(~inside_circle(p, C, Rs))[0]
            if idx.size > 0 and RESCUE_CAP_PER_STEP > 0:
                r.shuffle(idx)
                cand = idx[:min(RESCUE_CAP_PER_STEP, idx.size)]
                pull[cand] = -RESCUE_PULL * (p[cand] - C[None, :])

        v = 0.92 * v + eps + anchor + group_force + one_force + pull

        if np.any(outside) and OUTSIDE_DRAG > 0:
            v[outside] *= (1.0 - OUTSIDE_DRAG)

        sp = np.linalg.norm(v, axis=1, keepdims=True)
        v *= np.minimum(1.0, vmax / np.maximum(sp, 1e-12))

        p = p + v
        p, v = reflect_square(p, v, L_SIM)
        P[t] = p

    return P

P_div = simulate("diverse", seed=100)
P_grp = simulate("groups",  seed=101)
P_one = simulate("one",     seed=102)

# snapshots
snap = {
    "normal":  (P_div[t_form],  P_grp[t_form],  P_one[t_form],  False, None),
    "shock":   (P_div[t_shock], P_grp[t_shock], P_one[t_shock], True,  R0),
    "recovery":(P_div[t_end],   P_grp[t_end],   P_one[t_end],   True,  R1),
}

# ============================================================
# 5) Plot: 3 rows x 4 columns (left labels + 3 societies)
# ============================================================
TAB = plt.get_cmap("tab10")

def style_ax(ax):
    ax.set_aspect("equal")
    ax.set_xlim(-L_VIEW, L_VIEW)
    ax.set_ylim(-L_VIEW, L_VIEW)
    ax.set_xticks([]); ax.set_yticks([])
    for sp in ax.spines.values():
        sp.set_visible(False)

def add_circle(ax, radius):
    ax.add_patch(Circle(tuple(C), radius, fill=False, linewidth=2))

def outside_rate(p, radius):
    return (~inside_circle(p, C, radius)).mean()

def draw_div(ax, p):
    ax.scatter(p[:,0], p[:,1], s=18, marker="o", alpha=0.85,
               c=trait, cmap="viridis", vmin=0, vmax=1, linewidths=0)

def draw_groups(ax, p):
    for k in range(K_GROUPS):
        m = (z == k)
        if m.any():
            ax.scatter(p[m,0], p[m,1], s=18, marker="o", alpha=0.85,
                       color=TAB(k), linewidths=0)

def draw_one(ax, p):
    ax.scatter(p[:,0], p[:,1], s=18, marker="o", alpha=0.90,
               color=TAB(1), linewidths=0)

# ---- Text (all English) ----
TITLE = "Storyboard: Formation → Shock (safe zone) → Recovery"

row_labels = [
    "Normal times:\nshape of society",
    "Shock:\ninside the circle is safe",
    "Recovery:\nthe safe zone expands\nand people move inward",
]

# (改行つきで「重ならない」列タイトル)
col_titles = [
    "Diverse society",
    "Society with\nmultiple groups",
    "Fully unified\nsociety",
]

# (長文は2行にして、ショック右列の下に固定)
MESSAGE = "Outcome depends strongly\non where the group is located"

# ---- Figure & Grid ----
fig = plt.figure(figsize=(14.6, 8.6))
gs = fig.add_gridspec(
    3, 4,
    width_ratios=[2.45, 3.10, 3.10, 3.10],
    wspace=0.18,
    hspace=0.36
)

# Left column: row labels (no clipping)
for i in range(3):
    ax = fig.add_subplot(gs[i, 0])
    ax.axis("off")
    ax.text(0.02, 0.50, row_labels[i], ha="left", va="center",
            fontsize=18, color="#777777")

# Panels
row_keys = ["normal", "shock", "recovery"]
axs = [[None]*3 for _ in range(3)]
ax_unified_shock = None

for i, key in enumerate(row_keys):
    pd, pg, po, show_circle, rad = snap[key]
    for j, p in enumerate([pd, pg, po]):
        ax = fig.add_subplot(gs[i, j+1])
        axs[i][j] = ax
        style_ax(ax)

        if j == 0:
            draw_div(ax, p)
        elif j == 1:
            draw_groups(ax, p)
        else:
            draw_one(ax, p)

        if show_circle:
            add_circle(ax, rad)
            out = outside_rate(p, rad)
            ax.text(0.02, 0.02, f"Outside {out*100:.1f}%",
                    transform=ax.transAxes, ha="left", va="bottom",
                    fontsize=12, color="#888888",
                    bbox=dict(facecolor="white", edgecolor="none", alpha=0.65, pad=1.5))

        if (key == "shock") and (j == 2):
            ax_unified_shock = ax

# ---- Layout margins (reserve top area for titles) ----
fig.subplots_adjust(left=0.03, right=0.99, top=0.86, bottom=0.06)

# ---- Column titles placed with fig.text (so they NEVER overlap with suptitle) ----
# Use the 1st row axes positions
y_header = max(axs[0][0].get_position().y1,
               axs[0][1].get_position().y1,
               axs[0][2].get_position().y1) + 0.015

for j in range(3):
    pos = axs[0][j].get_position()
    x = pos.x0 + pos.width/2
    fig.text(x, y_header, col_titles[j],
             ha="center", va="bottom",
             fontsize=18, color="#777777")

# ---- Suptitle higher (no overlap) ----
fig.suptitle(TITLE, fontsize=20, y=0.965, color="#777777")

# ---- Message under the unified-shock panel (figure coords, no overlap) ----
if ax_unified_shock is not None:
    pos = ax_unified_shock.get_position()
    x = pos.x0 + pos.width/2
    y = pos.y0 - 0.035  # small gap below the shock panel
    fig.text(x, y, MESSAGE, ha="center", va="top",
             fontsize=15, color="#888888")

plt.show()

# Optional save:
# fig.savefig("storyboard_clean.png", dpi=300, bbox_inches="tight", pad_inches=0.02)

参考

Python3ではじめるシステムトレード【第2版】環境構築と売買戦略

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?