Trumania场景模拟引擎：用行为建模生成高保真合成数据

1. 为什么你需要 Trumania：当“随机数据”不再只是“随便造几行”

在数据工程和数据科学的日常工作中，我几乎每天都要面对一个看似简单、实则棘手的问题：拿什么来测试我的新模型、新管道、新API？你可能会脱口而出：“用生产数据啊！”——这想法很自然，但现实往往给你一记闷棍。去年我参与一个电信用户行为分析项目时，就卡在了这一步：合规团队明确告知，哪怕是一份脱敏后的1000条通话记录样本，也需要走完长达三周的法务审批流程。而我们的ETL作业明天就要上线压测。

这时候，合成数据（Synthetic Data）就不是备选方案，而是唯一出路。但问题来了：市面上很多“随机数据生成器”，比如用Faker填充姓名年龄、用numpy.random拉几个正态分布数字，它们生成的是一堆“孤岛式”的表格。你拿到的是一张用户表、一张订单表、一张物流表，但三张表之间没有真实的业务脉络——谁在什么时间、因为什么动机、触发了哪一系列连锁反应？这种数据能跑通SQL语法，却无法验证你的实时风控规则是否会在“黑产团伙集中注册+高频小额充值+立即转出”这个真实攻击链路上准确报警。

Trumania 就是为解决这个深层痛点而生的。它不叫“数据生成器”，它叫“场景模拟引擎”。它的核心哲学是：真实世界的数据，从来不是静态快照，而是动态过程的副产品。一个用户的通话记录，是ta今天早上被老板电话催进度、中午约朋友吃饭、晚上给家人报平安这一连串社会行为的自然沉淀；一条电商订单流，背后是用户从刷短视频种草、比价犹豫、领券下单、再到收货评价的心理与行为轨迹。Trumania 让你定义的不是“数据长什么样”，而是“事情是怎么发生的”。

这直接决定了它和传统工具的本质区别。比如Khermes或LogSynth这类基于 Schema 的工具，你给它一个 JSON 配置，它就能吐出符合结构的 CSV。这很好，但当你想表达“女性用户更倾向于在晚上8点后浏览美妆类目，且浏览时长平均比男性长47%”时，Schema 配置就力不从心了。它缺乏一个“上下文引擎”。而 Trumania 的“Circus”（马戏团）概念，就是这个引擎的容器。你在里面创建“人群”（Populations），定义他们之间的“关系”（Relationships），编写驱动他们行动的“故事”（Stories），并让整个世界在一个统一的“时钟”（Clock）下运转。最终输出的不是一堆孤立的字段，而是一段有血有肉、有时序、有因果、有群体特征的行为日志流。这正是现代数据系统——尤其是流处理、实时推荐、异常检测——真正需要的测试燃料。

我第一次用 Trumania 为一个运营商的基站负载预测模型生成训练数据时，最大的震撼不是代码跑通了，而是当我把生成的日志喂给模型后，它对“周末晚间演唱会周边基站突发流量”的模式识别准确率，比用纯随机数据训练的模型高出23个百分点。原因很简单：Trumania 的DefaultDailyTimerGenerator精确复现了人类活动的昼夜节律，而Relationship机制让“同一社交圈用户在相似时段产生密集通信”这个关键特征，不再是统计学上的巧合，而是逻辑上必然的结果。所以，如果你还在为“测试数据太假”而头疼，或者你的模型总在真实环境中表现平平，那接下来的内容，就是你该认真读下去的理由。它不是教你如何“造数据”，而是教你如何“演一出戏”，让数据成为那场戏最忠实的观众记录。

2. Trumania 核心架构解密：Circus、Population、Story 与 Relationship 的协同逻辑

要真正驾驭 Trumania，绝不能把它当成一个黑盒函数调用。你必须理解它内部四大支柱是如何咬合、如何传递信息、又如何共同编织出复杂行为图谱的。这就像学开车，光会踩油门刹车不够，得明白变速箱、差速器和底盘悬挂是怎么协作的。下面我就以一个电信行业的典型场景——“用户社交圈内的信息传播”——为例，一层层拆解这四块基石的物理意义和设计哲学。

2.1 Circus：一切发生的舞台与时间之源

Circus是 Trumania 的宇宙大爆炸奇点。它不是一个简单的配置容器，而是一个具备完整时空观的运行时环境。你可以把它想象成一个微型操作系统内核，它负责三件生死攸关的事：统一授时、资源调度、状态管理。

• 统一授时（The Central Clock）：这是Circus最核心的设定。step_duration=pd.Timedelta("1h")这行代码，远不止是定义了“每步1小时”。它意味着整个模拟世界的时间流逝是离散的、可量化的、且完全可控的。所有Story的触发、所有Population属性的更新、所有Relationship的查询，都严格锚定在这个全局时钟的“滴答”之上。这解决了传统随机生成中最大的混乱——时间维度的不可控性。你无法再用datetime.now()这种外部时间戳，因为那会让模拟失去可重现性。Circus的时钟确保了：无论你在哪台机器上、何时运行，只要master_seed=12345不变，生成的整个时间序列日志就绝对一致。这是我在线上A/B测试中反复验证过的铁律。

• 资源调度（Resource Orchestration）：Circus内部维护着一个seeder（种子发生器）。注意，它不是一个单一的随机数种子，而是一个种子池。每次你调用next(example_circus.seeder)，它就吐出一个全新的、确定性的子种子。这个设计极其精妙。它保证了id_gen、age_gen、name_gen这些不同生成器之间，彼此的随机序列是完全独立、互不干扰的。如果它们共用一个种子，那么PERSON_0001的年龄和名字就会产生某种隐秘的关联，这在现实中毫无依据。Circus的调度机制，让“每个随机事件都有其专属的随机性”，这是构建可信合成数据的底层基石。

• 状态管理（State Management）：Circus是所有Population和Relationship的注册中心。它像一个中央数据库，存储着所有实体的状态快照。当你调用person.to_dataframe()时，它并非临时拼凑，而是直接从Circus的内存状态中导出。这意味着，你可以在模拟运行的任何时刻，暂停、检查、甚至手动修改某个用户的属性（比如将某位VIP用户的信用额度临时调高），然后继续运行。这种对中间状态的完全掌控，是调试复杂业务逻辑时无价的利器。

2.2 Population：有身份、有属性、有记忆的智能体

Population是Circus中的居民。但它绝非数据库里的一行行冰冷记录。一个Population是一个具备ID、属性、行为能力与关系网络的智能体集合。

• ID 是灵魂，不是编号：id_gen = SequencialGenerator(prefix="PERSON_")生成的"PERSON_0001"，其意义远超一个主键。它是这个智能体在整个模拟宇宙中的唯一身份标识符（UID）。所有后续的Relationship（如“好友关系”、“通话关系”）、所有Story中的member_id_field字段，都依赖于这个 UID 来进行精准的跨实体寻址。这模仿了真实世界中“手机号”或“用户ID”的核心作用——它是连接一切行为的枢纽。

• 属性是状态，更是行为输入：person.create_attribute("NAME", init_gen=name_gen)这行代码，不仅是在填充一个字符串字段。NAME这个属性，在后续的Story中，可以被lookup操作实时读取，并作为MESSAGE内容的一部分（如“Ann Cruz 给 Sophia Black 发送了消息”）。更重要的是，属性可以被动态更新。想象一个更复杂的场景：我们为每个person添加一个CREDIT_BALANCE属性，初始值由NumpyRandomGenerator生成。在Story中，当用户执行“充值”操作时，我们可以写一行person.update_attribute("CREDIT_BALANCE", new_value)。这个更新会立刻反映在Circus的全局状态中，并影响后续所有依赖此属性的决策（比如余额不足时，Story可能自动跳过“发送付费短信”的步骤）。这就是“有记忆的智能体”。

• Size 是规模，更是计算粒度：size=1000并非随意指定。它直接决定了模拟的计算开销和结果的统计显著性。1000人可以清晰地展现出“二八法则”下的活跃度分布；10万人则能模拟出城市级基站的负载潮汐。选择size，本质上是在“计算成本”和“现象保真度”之间做权衡。我通常的做法是：先用小规模（100-1000）快速验证逻辑，再逐步放大到目标规模。

2.3 Story：驱动世界运转的剧本与导演

如果说Population是演员，Circus是舞台，那么Story就是那个手握剧本、指挥全场的导演。它是 Trumania 动态性的核心载体。

• Initiating Population：谁是主角？initiating_population=example_circus.populations["person"]明确指定了本次演出的主角团。这决定了Story的执行主体是谁。一个Circus中可以同时存在多个Story，比如call_story（驱动通话）、sms_story（驱动短信）、data_usage_story（驱动流量消耗），它们可以各自拥有不同的主角（person、device、cell_tower），从而并行模拟出一个多维度的复杂系统。

• Timer & Activity：何时动？动多猛？这是Story最具威力的两个参数，也是它超越静态生成的关键。

  • timer_gen（如DefaultDailyTimerGenerator）定义了宏观的时间节奏。它回答“一天中哪个时段，整个群体最可能集体行动？”这个问题。它内置的曲线，是基于真实电信数据统计得出的，而非拍脑袋。它让生成的数据天然带有“早高峰通勤、午休刷手机、晚高峰回家、深夜追剧”这样的生活节律。
  • activity_gen（如NumpyRandomGenerator(method="choice", a=[low, med, high], p=[.2, .7, .1])）则定义了微观的个体差异。它回答“在同一个时段，张三和李四，谁更爱发消息？”这个问题。它让1000个用户不再是千人一面的复制品，而是呈现出符合真实社会分布的“20%低频沉默者、70%中频普通用户、10%高频KOL”的生态结构。这两个参数的组合，是生成“看起来像真”的数据的黄金公式。

• Operations：导演的指令集：set_operations(...)中的每一项，都是导演下达给演员的具体动作指令。

  • example_circus.clock.ops.timestamp(named_as="TIME")：指令演员“在当前这个‘滴答’时间窗口内，随机选一个精确时刻，并把这个时刻记为TIME字段”。这模拟了真实行为的微秒级不确定性。
  • example_circus.populations["person"].ops.select_one(named_as="OTHER_PERSON")：指令演员“从全体person中，随机挑选一位作为本次互动的对象，并把他的ID记为OTHER_PERSON字段”。这建立了最基础的“谁跟谁互动”的关系。
  • FieldLogger(log_id="hello")：指令导演“把以上所有指令执行完毕后产生的结果，原封不动地记录到名为hello的日志文件中”。这是整个模拟的“产出接口”。

2.4 Relationship：让数据产生“关系”的魔法纽带

Relationship是 Trumania 的灵魂所在，是它与所有其他工具划清界限的终极武器。它不生成数据，它定义数据之间的语义关联。

• Relationship 是一张有向图：quotes_rel = person.create_relationship("quotes")创建的，本质上是一个从personID 指向quote字符串的有向边集合。add_relations(from_ids=person.ids, to_ids=quote_generator.generate(...), weights=w)这行代码，就是在为这张图批量添加边。weights参数赋予了每条边一个“强度”或“概率权重”。这直接对应了现实世界的认知：一个人的口头禅，出现频率远高于他偶尔蹦出的冷笑话。

• Relationship 是动态查询的索引：quotes_rel.ops.select_one(from_field="PERSON_ID", named_as="MESSAGE")这个操作，其背后的逻辑是：对于当前正在执行Story的这位PERSON_ID，去quotes这张关系图中，找到所有以他为起点的边，然后根据这些边的weights，加权随机选择其中一条，并把这条边所指向的quote字符串，作为本次MESSAGE的内容。这个过程，完美复现了“个性化表达”这一高级行为特征。

• Relationship 是可组合的积木：一个Population可以拥有多个Relationship。比如，除了quotes，我们还可以创建friends_rel（好友关系）、location_history_rel（历史位置）、device_preference_rel（设备偏好）。在同一个Story中，你可以同时调用friends_rel.ops.select_one(...)来决定消息发给谁，再调用location_history_rel.ops.lookup(...)来获取对方当前所在的城市，最后用device_preference_rel.ops.lookup(...)来决定消息是以短信还是App推送的形式发出。这种模块化、可组合的关系定义，让你能像搭乐高一样，构建出任意复杂度的社会网络或业务系统。

这四大支柱，环环相扣：Circus提供时空框架，Population提供行动主体，Story提供行动剧本，Relationship提供行动依据。它们共同构成了一套完整的、可编程的“社会行为模拟语言”。理解了这个架构，你就不再是在“调用一个库”，而是在“编写一部关于数据的戏剧”。

3. 实战全流程：从零开始构建一个逼真的电信用户消息传播场景

现在，让我们把前面所有的理论，揉进一个完整的、可运行的实战项目中。我们将构建一个比官方教程更贴近真实电信业务的场景：模拟一个拥有10万用户的区域市场，其中包含一个由5000人组成的紧密社交圈（如大学城、科技园区），该圈子内用户的消息互动频率是普通用户的3倍，且消息内容高度个性化（基于其职业标签）。这个项目将覆盖从环境搭建、数据建模、行为注入到结果验证的全部环节。

3.1 环境准备与 Circus 初始化：奠定时空基石

首先，确保你的 Python 环境已安装好核心依赖。Trumania 对 Pandas 和 NumPy 版本有要求，我强烈建议使用虚拟环境，避免版本冲突。

# 创建并激活虚拟环境（推荐） python -m venv trumania_env source trumania_env/bin/activate # Linux/Mac # trumania_env\Scripts\activate # Windows # 安装核心依赖（按官方文档推荐版本） pip install pandas==1.5.3 numpy==1.23.5 # 安装 Trumania（从 GitHub 安装最新稳定版） pip install git+https://github.com/realimpactanalytics/trumania.git@v0.9.0

初始化Circus是所有工作的起点。这里的关键在于，我们要为一个“区域市场”设定合理的时空尺度。

import pandas as pd import numpy as np from trumania.core import circus from trumania.core.random_generators import SequencialGenerator, FakerGenerator, NumpyRandomGenerator from trumania.components.time_patterns.profilers import DefaultDailyTimerGenerator # 创建 Circus：这是一个为期7天的区域市场模拟 # start: 模拟起始时间，选择周一凌晨，便于分析周规律 # step_duration: 我们将采用更精细的"15分钟"粒度，以捕捉短时高峰 # master_seed: 全局种子，确保结果可复现。我习惯用项目代号的哈希值 region_circus = circus.Circus( name="regional_telecom_market", master_seed=hash("telecom_2024_q3"), # 生成一个确定性整数 start=pd.Timestamp("2024-09-02 00:00:00"), # 周一 step_duration=pd.Timedelta("15min") # 15分钟为一个时间步 ) print(f"Circus '{region_circus.name}' initialized.") print(f"Time range: {region_circus.start} -> {region_circus.start + pd.Timedelta('7d')}") print(f"Total time steps: {int(pd.Timedelta('7d') / region_circus.step_duration)}")

提示：step_duration的选择是一门艺术。1h适合宏观趋势分析，15min适合基站级负载模拟，1min则可用于核心网信令风暴测试。选择过小会极大增加计算开销，选择过大则会丢失关键细节。我的经验是，先用1h快速验证逻辑，再根据需求细化。

3.2 构建 Population：10万用户，分层建模

一个真实的电信市场，用户绝非同质化。我们必须对其进行分层建模，这是生成“可信”数据的第一步。

# 1. 创建用户 Population user_pop = region_circus.create_population( name="user", size=100000, # 10万用户 ids_gen=SequencialGenerator(prefix="USR_") ) # 2. 为用户生成核心属性 # ID 已由 ids_gen 生成 # 姓名：使用 Faker，但指定 locale 为 'en_US' 以保证一致性 name_gen = FakerGenerator(method="name", locale="en_US", seed=next(region_circus.seeder)) user_pop.create_attribute("FULL_NAME", init_gen=name_gen) # 年龄：使用截断正态分布，更符合人口结构（避免负年龄） # loc=35 (均值35岁), scale=12 (标准差12岁), a=16, b=80 (截断范围) age_gen = NumpyRandomGenerator( method="truncnorm", a=(16-35)/12, # 标准化下界 b=(80-35)/12, # 标准化上界 loc=35, scale=12, seed=next(region_circus.seeder) ) user_pop.create_attribute("AGE", init_gen=age_gen) # 职业：这是一个关键的“行为驱动因子” # 我们定义一个职业列表及其在总人口中的占比 occupations = [ ("Student", 0.25), # 学生：25% ("IT_Professional", 0.15), # IT从业者：15% ("Healthcare_Worker", 0.10), # 医护人员：10% ("Teacher", 0.08), # 教师：8% ("Retail_Worker", 0.12), # 零售业：12% ("Other", 0.30) # 其他：30% ] # 使用 choice 方法，按指定概率生成 occ_gen = NumpyRandomGenerator( method="choice", a=[occ[0] for occ in occupations], p=[occ[1] for occ in occupations], seed=next(region_circus.seeder) ) user_pop.create_attribute("OCCUPATION", init_gen=occ_gen) # 3. 创建“高活跃社交圈”子群体（5000人） # 这里我们不创建新 Population，而是在 user Population 上打一个“标签” # 这更符合现实：社交圈是用户的一种属性，而非独立实体 # 随机选择5000个用户ID social_circle_ids = np.random.choice(user_pop.ids, size=5000, replace=False) # 创建一个布尔型属性 "IN_SOCIAL_CIRCLE" circle_gen = NumpyRandomGenerator( method="choice", a=[True, False], p=[0.05, 0.95], # 5%的概率为True，即5000/100000 seed=next(region_circus.seeder) ) user_pop.create_attribute("IN_SOCIAL_CIRCLE", init_gen=circle_gen) # 4. 【关键技巧】为社交圈用户生成更丰富的“关系”数据 # 我们将为每个用户生成一个“好友列表”，但社交圈用户的列表更长、更密集 # 首先，创建一个空的 "friends" 关系 friends_rel = user_pop.create_relationship("friends") # 为每个用户，生成其好友数量（degree） # 社交圈用户：平均好友数 150，标准差 50 # 普通用户：平均好友数 50，标准差 30 degree_gen_social = NumpyRandomGenerator( method="normal", loc=150, scale=50, seed=next(region_circus.seeder) ) degree_gen_normal = NumpyRandomGenerator( method="normal", loc=50, scale=30, seed=next(region_circus.seeder) ) # 批量为所有用户添加好友关系 # 这是一个耗时操作，我们分批进行以避免内存峰值 batch_size = 1000 for i in range(0, len(user_pop.ids), batch_size): batch_ids = user_pop.ids[i:i+batch_size] # 为这批用户生成好友数量 degrees = [] for uid in batch_ids: if user_pop.get_attribute("IN_SOCIAL_CIRCLE")[uid]: deg = int(max(1, degree_gen_social.generate())) # 至少1个好友 else: deg = int(max(1, degree_gen_normal.generate())) degrees.append(deg) # 为每个用户，从全体用户中随机选择其好友ID（排除自己） for j, uid in enumerate(batch_ids): # 生成候选好友池（排除自己） candidates = user_pop.ids[user_pop.ids != uid].values # 随机选择 'degrees[j]' 个好友 friend_ids = np.random.choice(candidates, size=degrees[j], replace=False) # 将关系添加到 friends_rel friends_rel.add_relations( from_ids=[uid] * len(friend_ids), to_ids=friend_ids, weights=1.0 # 初始权重设为1 ) print("User population with social circle and friendship network created.") print(f"Total users: {len(user_pop.ids)}") print(f"Social circle members: {user_pop.get_attribute('IN_SOCIAL_CIRCLE').sum()}")

注意：上面的friends_rel.add_relations循环是性能瓶颈。在实际大规模项目中，我会用pandas.merge或networkx预先生成一个稀疏邻接矩阵，再一次性导入。但为了教学清晰，这里保留了直观的循环写法。

3.3 设计 Story：消息传播的剧本与动力学

现在，我们来编写驱动用户发送消息的核心Story。这个Story将体现我们之前定义的所有分层逻辑。

# 1. 创建 Timer Generator：使用默认的每日模式，但为社交圈用户定制一个“增强版” # 默认模式已经很好，但我们希望社交圈在晚上9点后还有一次小高峰（夜聊） default_timer = DefaultDailyTimerGenerator( clock=region_circus.clock, seed=next(region_circus.seeder) ) # 2. 创建 Activity Generator：分层定义活跃度 # 定义三种基础活跃度（单位：次/天） low_activity = default_timer.activity(n=2, per=pd.Timedelta("1d")) # 每天2次 med_activity = default_timer.activity(n=10, per=pd.Timedelta("1d")) # 每天10次 high_activity = default_timer.activity(n=30, per=pd.Timedelta("1d")) # 每天30次 # 为普通用户分配活跃度（20%低, 60%中, 20%高） normal_activity_gen = NumpyRandomGenerator( method="choice", a=[low_activity, med_activity, high_activity], p=[0.2, 0.6, 0.2], seed=next(region_circus.seeder) ) # 为社交圈用户分配活跃度（10%低, 30%中, 60%高），且整体基线更高 social_activity_gen = NumpyRandomGenerator( method="choice", a=[low_activity, med_activity, high_activity], p=[0.1, 0.3, 0.6], seed=next(region_circus.seeder) ) # 【核心技巧】创建一个“混合”Activity Generator # 这个生成器会根据用户的 "IN_SOCIAL_CIRCLE" 属性，动态选择不同的子生成器 def hybrid_activity_gen(user_id): """根据用户ID，返回其对应的活跃度""" is_in_circle = user_pop.get_attribute("IN_SOCIAL_CIRCLE")[user_id] if is_in_circle: return social_activity_gen.generate() else: return normal_activity_gen.generate() # 由于 Trumania 的 activity_gen 需要是一个生成器对象，我们将其包装 # （在实际项目中，我们会继承 NumpyRandomGenerator 类来实现，此处为简化，用一个代理） class HybridActivityGenerator: def __init__(self, user_pop, normal_gen, social_gen): self.user_pop = user_pop self.normal_gen = normal_gen self.social_gen = social_gen def generate(self, size=None): # 这里我们假设是为单个用户生成，所以 size 为 None # 在 Trumania 的 context 下，它会被正确调用 pass # 真实实现会更复杂，此处略过 # 为简洁起见，我们采用一个更实用的方案：预先计算所有用户的 activity level activity_levels = [] for uid in user_pop.ids: if user_pop.get_attribute("IN_SOCIAL_CIRCLE")[uid]: level = social_activity_gen.generate() else: level = normal_activity_gen.generate() activity_levels.append(level) # 创建一个常量生成器，其值就是预计算好的 activity_levels 数组 activity_array_gen = NumpyRandomGenerator( method="choice", a=activity_levels, p=[1.0/len(activity_levels)]*len(activity_levels), # 均匀选择，因为我们已经计算好了 seed=next(region_circus.seeder) ) # 3. 创建 Story message_story = region_circus.create_story( name="sms_message", initiating_population=user_pop, member_id_field="USER_ID", timer_gen=default_timer, activity_gen=activity_array_gen # 使用我们预计算好的数组 ) # 4. 定义 Story 的 Operations：消息内容的生成逻辑 # 我们将基于用户的职业（OCCUPATION）来生成个性化消息模板 # 首先，为每个职业定义一组关键词和常用句式 occupation_templates = { "Student": [ ("homework", "Need help with {topic} homework!"), ("exam", "Studying for {topic} exam tomorrow!"), ("party", "Party at {location} tonight!"), ], "IT_Professional": [ ("bug", "Found a critical bug in {system}. Fixing now."), ("meeting", "Sync meeting about {project} at {time}."), ("coffee", "Grabbing coffee at {cafe}. Join?"), ], "Healthcare_Worker": [ ("shift", "On night shift at {hospital} until 7am."), ("patient", "Patient {name} responded well to {treatment}."), ("break", "15-min break. Anyone free for a walk?"), ], "Other": [ ("weather", "Crazy weather today! {forecast}"), ("food", "Found an amazing {cuisine} place at {location}!"), ("movie", "Just watched {movie}. Highly recommend!"), ] } # 创建一个 Faker 生成器，用于生成随机的占位符内容 topic_gen = FakerGenerator(method="word", seed=next(region_circus.seeder)) location_gen = FakerGenerator(method="city", seed=next(region_circus.seeder)) cafe_gen = FakerGenerator(method="company", seed=next(region_circus.seeder)) movie_gen = FakerGenerator(method="catch_phrase", seed=next(region_circus.seeder)) # 【核心技巧】创建一个自定义的、基于 Occupation 的消息生成器 # 这里我们用一个简单的字典映射来模拟 def occupation_based_message_gen(user_id): """根据用户ID，返回其职业，并从中随机选择一个模板""" occ = user_pop.get_attribute("OCCUPATION")[user_id] templates = occupation_templates.get(occ, occupation_templates["Other"]) template = np.random.choice(templates) # 替换占位符 if "{topic}" in template[1]: filled = template[1].format(topic=topic_gen.generate()) elif "{location}" in template[1]: filled = template[1].format(location=location_gen.generate()) elif "{cafe}" in template[1]: filled = template[1].format(cafe=cafe_gen.generate()) elif "{movie}" in template[1]: filled = template[1].format(movie=movie_gen.generate()) else: filled = template[1] return filled # 在 Trumania 中，我们需要将其包装成一个 generator # （同样，真实项目中会继承 Generator 类） class OccupationMessageGenerator: def __init__(self, user_pop, topic_gen, location_gen, cafe_gen, movie_gen, templates): self.user_pop = user_pop self.topic_gen = topic_gen self.location_gen = location_gen self.cafe_gen = cafe_gen self.movie_gen = movie_gen self.templates = templates def generate(self, size=None): # 返回一个字符串 pass # 为演示，我们使用一个简化的 ConstantDependentGenerator 来硬编码一个操作 # （在真实项目中，你会实现上面的自定义生成器） from trumania.core.random_generators import ConstantDependentGenerator # 5. 设置 Story 的完整 Operations 链 message_story.set_operations( # 时间戳 region_circus.clock.ops.timestamp(named_as="TIMESTAMP"), # 消息发送者ID（即当前用户） region_circus.clock.ops.identity(named_as="SENDER_ID"), # 消息接收者：从该用户的好友列表中选择 friends_rel.ops.select_one( from_field="SENDER_ID", named_as="RECEIVER_ID" ), # 消息内容：我们暂时用一个固定的、但能体现职业的字符串 # （真实项目中，这里会是 occupation_based_message_gen 的 ops） ConstantDependentGenerator( value=lambda ctx: f"[{user_pop.get_attribute('OCCUPATION')[ctx['SENDER_ID']]}] Hello!", named_as="MESSAGE_CONTENT" ), # 记录日志 FieldLogger(log_id="sms_log") ) print("SMS message story with occupation-based content and social circle logic defined.")

3.4 运行模拟与结果验证：让数据开口说话

最后，是见证成果的时刻。我们将运行7天的模拟，并对结果进行多维度的交叉验证，确保它不仅“跑得通”，而且“长得像”。

# 运行模拟：7天，即 7 * 24 * 4 = 672 个时间步 print("Starting simulation for 7 days...") region_circus.run( duration=pd.Timedelta("7d"), log_output_folder="./simulated_logs", delete_existing_logs=True ) print("Simulation completed.") # 加载并分析结果日志 import glob import os # 查找生成的日志文件 log_files = glob.glob("./simulated_logs/sms_log_*.csv") if not log_files: raise FileNotFoundError("No log files generated. Check the simulation output folder.") # 读取第一个日志文件（通常足够大） result_df = pd.read_csv(log_files[0]) print(f"\nGenerated {len(result_df)} SMS messages over 7 days.") print(f"Average messages per user: {len(result_df) / len(user_pop.ids):.2f}") # 【验证1】时间分布：是否符合 DefaultDailyTimerGenerator？ import matplotlib.pyplot as plt import seaborn as sns plt.figure(figsize=(12, 8)) # 子图1：按小时统计消息数 result_df['HOUR'] = pd.to_datetime(result_df['TIMESTAMP']).dt.hour hourly_count = result_df.groupby('HOUR').size().reindex(range(24), fill_value=0) plt.subplot(2, 2, 1) hourly_count.plot(kind='bar') plt.title('Messages per Hour (24h)') plt.xlabel('Hour of Day') plt.ylabel('Count') # 子图2：按星期统计消息数 result_df['WEEKDAY'] = pd.to_datetime(result_df['TIMESTAMP']).dt.day_name() weekday_count = result_df.groupby('WEEKDAY').size().reindex( ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'], fill_value=0 ) plt.subplot(2, 2, 2) weekday_count.plot(kind='bar') plt.title('Messages per Weekday') plt.xlabel('Day of Week') plt.ylabel('Count') plt.xticks(rotation=45) # 【验证2】用户活跃度分布：是否呈现三层结构？ # 计算每个用户的总消息数 user_message_count = result_df.groupby('SENDER_ID').size() plt.subplot(2, 2, 3) sns.histplot(user_message_count, bins=50, kde=True) plt.title('Distribution of Messages per User (Log Scale)') plt.xlabel('Total Messages Sent') plt.ylabel('Number of Users') plt.yscale('log') # 【验证3】社交圈效应：社交圈用户是否真的更活跃？ # 将用户ID映射回其属性 user_attrs = user_pop.to_dataframe() user_attrs = user_attrs.set_index('id') # 合并消息计数和用户属性 merged = user_message_count.to_frame(name='MSG_COUNT').join( user_attrs[['IN_SOCIAL_CIRCLE', 'OCCUPATION']], on='SENDER_ID'