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_durationpd.Timedelta(1h)这行代码远不止是定义了“每步1小时”。它意味着整个模拟世界的时间流逝是离散的、可量化的、且完全可控的。所有Story的触发、所有Population属性的更新、所有Relationship的查询都严格锚定在这个全局时钟的“滴答”之上。这解决了传统随机生成中最大的混乱——时间维度的不可控性。你无法再用datetime.now()这种外部时间戳因为那会让模拟失去可重现性。Circus的时钟确保了无论你在哪台机器上、何时运行只要master_seed12345不变生成的整个时间序列日志就绝对一致。这是我在线上A/B测试中反复验证过的铁律。资源调度Resource OrchestrationCircus内部维护着一个seeder种子发生器。注意它不是一个单一的随机数种子而是一个种子池。每次你调用next(example_circus.seeder)它就吐出一个全新的、确定性的子种子。这个设计极其精妙。它保证了id_gen、age_gen、name_gen这些不同生成器之间彼此的随机序列是完全独立、互不干扰的。如果它们共用一个种子那么PERSON_0001的年龄和名字就会产生某种隐秘的关联这在现实中毫无依据。Circus的调度机制让“每个随机事件都有其专属的随机性”这是构建可信合成数据的底层基石。状态管理State ManagementCircus是所有Population和Relationship的注册中心。它像一个中央数据库存储着所有实体的状态快照。当你调用person.to_dataframe()时它并非临时拼凑而是直接从Circus的内存状态中导出。这意味着你可以在模拟运行的任何时刻暂停、检查、甚至手动修改某个用户的属性比如将某位VIP用户的信用额度临时调高然后继续运行。这种对中间状态的完全掌控是调试复杂业务逻辑时无价的利器。2.2 Population有身份、有属性、有记忆的智能体Population是Circus中的居民。但它绝非数据库里的一行行冰冷记录。一个Population是一个具备ID、属性、行为能力与关系网络的智能体集合。ID 是灵魂不是编号id_gen SequencialGenerator(prefixPERSON_)生成的PERSON_0001其意义远超一个主键。它是这个智能体在整个模拟宇宙中的唯一身份标识符UID。所有后续的Relationship如“好友关系”、“通话关系”、所有Story中的member_id_field字段都依赖于这个 UID 来进行精准的跨实体寻址。这模仿了真实世界中“手机号”或“用户ID”的核心作用——它是连接一切行为的枢纽。属性是状态更是行为输入person.create_attribute(NAME, init_genname_gen)这行代码不仅是在填充一个字符串字段。NAME这个属性在后续的Story中可以被lookup操作实时读取并作为MESSAGE内容的一部分如“Ann Cruz 给 Sophia Black 发送了消息”。更重要的是属性可以被动态更新。想象一个更复杂的场景我们为每个person添加一个CREDIT_BALANCE属性初始值由NumpyRandomGenerator生成。在Story中当用户执行“充值”操作时我们可以写一行person.update_attribute(CREDIT_BALANCE, new_value)。这个更新会立刻反映在Circus的全局状态中并影响后续所有依赖此属性的决策比如余额不足时Story可能自动跳过“发送付费短信”的步骤。这就是“有记忆的智能体”。Size 是规模更是计算粒度size1000并非随意指定。它直接决定了模拟的计算开销和结果的统计显著性。1000人可以清晰地展现出“二八法则”下的活跃度分布10万人则能模拟出城市级基站的负载潮汐。选择size本质上是在“计算成本”和“现象保真度”之间做权衡。我通常的做法是先用小规模100-1000快速验证逻辑再逐步放大到目标规模。2.3 Story驱动世界运转的剧本与导演如果说Population是演员Circus是舞台那么Story就是那个手握剧本、指挥全场的导演。它是 Trumania 动态性的核心载体。Initiating Population谁是主角initiating_populationexample_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(methodchoice, a[low, med, high], p[.2, .7, .1])则定义了微观的个体差异。它回答“在同一个时段张三和李四谁更爱发消息”这个问题。它让1000个用户不再是千人一面的复制品而是呈现出符合真实社会分布的“20%低频沉默者、70%中频普通用户、10%高频KOL”的生态结构。这两个参数的组合是生成“看起来像真”的数据的黄金公式。Operations导演的指令集set_operations(...)中的每一项都是导演下达给演员的具体动作指令。example_circus.clock.ops.timestamp(named_asTIME)指令演员“在当前这个‘滴答’时间窗口内随机选一个精确时刻并把这个时刻记为TIME字段”。这模拟了真实行为的微秒级不确定性。example_circus.populations[person].ops.select_one(named_asOTHER_PERSON)指令演员“从全体person中随机挑选一位作为本次互动的对象并把他的ID记为OTHER_PERSON字段”。这建立了最基础的“谁跟谁互动”的关系。FieldLogger(log_idhello)指令导演“把以上所有指令执行完毕后产生的结果原封不动地记录到名为hello的日志文件中”。这是整个模拟的“产出接口”。2.4 Relationship让数据产生“关系”的魔法纽带Relationship是 Trumania 的灵魂所在是它与所有其他工具划清界限的终极武器。它不生成数据它定义数据之间的语义关联。Relationship 是一张有向图quotes_rel person.create_relationship(quotes)创建的本质上是一个从personID 指向quote字符串的有向边集合。add_relations(from_idsperson.ids, to_idsquote_generator.generate(...), weightsw)这行代码就是在为这张图批量添加边。weights参数赋予了每条边一个“强度”或“概率权重”。这直接对应了现实世界的认知一个人的口头禅出现频率远高于他偶尔蹦出的冷笑话。Relationship 是动态查询的索引quotes_rel.ops.select_one(from_fieldPERSON_ID, named_asMESSAGE)这个操作其背后的逻辑是对于当前正在执行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 pandas1.5.3 numpy1.23.5 # 安装 Trumania从 GitHub 安装最新稳定版 pip install githttps://github.com/realimpactanalytics/trumania.gitv0.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( nameregional_telecom_market, master_seedhash(telecom_2024_q3), # 生成一个确定性整数 startpd.Timestamp(2024-09-02 00:00:00), # 周一 step_durationpd.Timedelta(15min) # 15分钟为一个时间步 ) print(fCircus {region_circus.name} initialized.) print(fTime range: {region_circus.start} - {region_circus.start pd.Timedelta(7d)}) print(fTotal time steps: {int(pd.Timedelta(7d) / region_circus.step_duration)})提示step_duration的选择是一门艺术。1h适合宏观趋势分析15min适合基站级负载模拟1min则可用于核心网信令风暴测试。选择过小会极大增加计算开销选择过大则会丢失关键细节。我的经验是先用1h快速验证逻辑再根据需求细化。3.2 构建 Population10万用户分层建模一个真实的电信市场用户绝非同质化。我们必须对其进行分层建模这是生成“可信”数据的第一步。# 1. 创建用户 Population user_pop region_circus.create_population( nameuser, size100000, # 10万用户 ids_genSequencialGenerator(prefixUSR_) ) # 2. 为用户生成核心属性 # ID 已由 ids_gen 生成 # 姓名使用 Faker但指定 locale 为 en_US 以保证一致性 name_gen FakerGenerator(methodname, localeen_US, seednext(region_circus.seeder)) user_pop.create_attribute(FULL_NAME, init_genname_gen) # 年龄使用截断正态分布更符合人口结构避免负年龄 # loc35 (均值35岁), scale12 (标准差12岁), a16, b80 (截断范围) age_gen NumpyRandomGenerator( methodtruncnorm, a(16-35)/12, # 标准化下界 b(80-35)/12, # 标准化上界 loc35, scale12, seednext(region_circus.seeder) ) user_pop.create_attribute(AGE, init_genage_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( methodchoice, a[occ[0] for occ in occupations], p[occ[1] for occ in occupations], seednext(region_circus.seeder) ) user_pop.create_attribute(OCCUPATION, init_genocc_gen) # 3. 创建“高活跃社交圈”子群体5000人 # 这里我们不创建新 Population而是在 user Population 上打一个“标签” # 这更符合现实社交圈是用户的一种属性而非独立实体 # 随机选择5000个用户ID social_circle_ids np.random.choice(user_pop.ids, size5000, replaceFalse) # 创建一个布尔型属性 IN_SOCIAL_CIRCLE circle_gen NumpyRandomGenerator( methodchoice, a[True, False], p[0.05, 0.95], # 5%的概率为True即5000/100000 seednext(region_circus.seeder) ) user_pop.create_attribute(IN_SOCIAL_CIRCLE, init_gencircle_gen) # 4. 【关键技巧】为社交圈用户生成更丰富的“关系”数据 # 我们将为每个用户生成一个“好友列表”但社交圈用户的列表更长、更密集 # 首先创建一个空的 friends 关系 friends_rel user_pop.create_relationship(friends) # 为每个用户生成其好友数量degree # 社交圈用户平均好友数 150标准差 50 # 普通用户平均好友数 50标准差 30 degree_gen_social NumpyRandomGenerator( methodnormal, loc150, scale50, seednext(region_circus.seeder) ) degree_gen_normal NumpyRandomGenerator( methodnormal, loc50, scale30, seednext(region_circus.seeder) ) # 批量为所有用户添加好友关系 # 这是一个耗时操作我们分批进行以避免内存峰值 batch_size 1000 for i in range(0, len(user_pop.ids), batch_size): batch_ids user_pop.ids[i:ibatch_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, sizedegrees[j], replaceFalse) # 将关系添加到 friends_rel friends_rel.add_relations( from_ids[uid] * len(friend_ids), to_idsfriend_ids, weights1.0 # 初始权重设为1 ) print(User population with social circle and friendship network created.) print(fTotal users: {len(user_pop.ids)}) print(fSocial 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( clockregion_circus.clock, seednext(region_circus.seeder) ) # 2. 创建 Activity Generator分层定义活跃度 # 定义三种基础活跃度单位次/天 low_activity default_timer.activity(n2, perpd.Timedelta(1d)) # 每天2次 med_activity default_timer.activity(n10, perpd.Timedelta(1d)) # 每天10次 high_activity default_timer.activity(n30, perpd.Timedelta(1d)) # 每天30次 # 为普通用户分配活跃度20%低, 60%中, 20%高 normal_activity_gen NumpyRandomGenerator( methodchoice, a[low_activity, med_activity, high_activity], p[0.2, 0.6, 0.2], seednext(region_circus.seeder) ) # 为社交圈用户分配活跃度10%低, 30%中, 60%高且整体基线更高 social_activity_gen NumpyRandomGenerator( methodchoice, a[low_activity, med_activity, high_activity], p[0.1, 0.3, 0.6], seednext(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, sizeNone): # 这里我们假设是为单个用户生成所以 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( methodchoice, aactivity_levels, p[1.0/len(activity_levels)]*len(activity_levels), # 均匀选择因为我们已经计算好了 seednext(region_circus.seeder) ) # 3. 创建 Story message_story region_circus.create_story( namesms_message, initiating_populationuser_pop, member_id_fieldUSER_ID, timer_gendefault_timer, activity_genactivity_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(methodword, seednext(region_circus.seeder)) location_gen FakerGenerator(methodcity, seednext(region_circus.seeder)) cafe_gen FakerGenerator(methodcompany, seednext(region_circus.seeder)) movie_gen FakerGenerator(methodcatch_phrase, seednext(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(topictopic_gen.generate()) elif {location} in template[1]: filled template[1].format(locationlocation_gen.generate()) elif {cafe} in template[1]: filled template[1].format(cafecafe_gen.generate()) elif {movie} in template[1]: filled template[1].format(moviemovie_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, sizeNone): # 返回一个字符串 pass # 为演示我们使用一个简化的 ConstantDependentGenerator 来硬编码一个操作 # 在真实项目中你会实现上面的自定义生成器 from trumania.core.random_generators import ConstantDependentGenerator # 5. 设置 Story 的完整 Operations 链 message_story.set_operations( # 时间戳 region_circus.clock.ops.timestamp(named_asTIMESTAMP), # 消息发送者ID即当前用户 region_circus.clock.ops.identity(named_asSENDER_ID), # 消息接收者从该用户的好友列表中选择 friends_rel.ops.select_one( from_fieldSENDER_ID, named_asRECEIVER_ID ), # 消息内容我们暂时用一个固定的、但能体现职业的字符串 # 真实项目中这里会是 occupation_based_message_gen 的 ops ConstantDependentGenerator( valuelambda ctx: f[{user_pop.get_attribute(OCCUPATION)[ctx[SENDER_ID]]}] Hello!, named_asMESSAGE_CONTENT ), # 记录日志 FieldLogger(log_idsms_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( durationpd.Timedelta(7d), log_output_folder./simulated_logs, delete_existing_logsTrue ) 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(fAverage 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_value0) plt.subplot(2, 2, 1) hourly_count.plot(kindbar) 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_value0 ) plt.subplot(2, 2, 2) weekday_count.plot(kindbar) plt.title(Messages per Weekday) plt.xlabel(Day of Week) plt.ylabel(Count) plt.xticks(rotation45) # 【验证2】用户活跃度分布是否呈现三层结构 # 计算每个用户的总消息数 user_message_count result_df.groupby(SENDER_ID).size() plt.subplot(2, 2, 3) sns.histplot(user_message_count, bins50, kdeTrue) 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(nameMSG_COUNT).join( user_attrs[[IN_SOCIAL_CIRCLE, OCCUPATION]], onSENDER_ID