> 持续流程改进 > 费用管理 > Ramp > 数据模板

您的费用管理数据模板

Ramp

← 返回改进指南

费用管理属性 (19) 费用管理活动 (12) 提取指南 (3)

您的费用管理数据模板

此模板为收集必要数据、分析并优化费用管理流程提供清晰路径。模板列出了必需属性、需要跟踪的关键活动，以及从Ramp提取数据的实用指南。使用此资源，确保收集流程发现和改进所需的完整信息。

建议收集的属性
需要跟踪的关键活动
Ramp数据提取指南

刚接触事件日志？了解如何创建流程挖掘事件日志.

下载模板

费用管理属性

以下是应纳入事件日志的数据字段，用于全面分析费用管理流程。

3 必需 6 建议 10 可选

	名称	说明
	事件时间戳 EventTimestamp	活动发生的准确日期和时间。
说明流程中的每项活动都有对应时间戳，用于记录其发生时间。这些时间数据用于按时间顺序排列事件，也是所有时间相关分析的基础。在流程挖掘中，时间戳用于计算活动之间的周期时间、衡量案例总时长和识别延迟。这些信息对于绩效监控和发现流程改进机会至关重要。为什么重要该属性提供事件的时间顺序，是所有时长计算和绩效分析不可或缺的依据。获取位置通常可在Ramp的事件日志或交易数据中，与活动记录或状态记录一同找到。示例 2023-10-26T10:00:00Z2023-10-26T14:30:00Z2023-10-27T09:00:00Z
	活动名称 ActivityName	费用管理流程中某个时间点发生的具体事件或任务的名称。
说明此属性描述费用报告生命周期中的单个步骤，例如'Expense Submitted'、'Manager Approved'或'Reimbursement Executed'。这些活动构成流程图中的节点，可用于可视化和分析流程顺序。分析活动有助于识别最常发生的步骤、瓶颈出现的位置，以及不同工作流之间的差异。它是理解操作顺序并衡量各阶段绩效的核心组成部分。为什么重要它定义流程图中的步骤，使您能够可视化和分析从开始到结束的流程顺序。获取位置通常从Ramp中与每份费用报告关联的事件日志或状态变更记录中提取。示例已提交费用经理已批准待财务审核已执行报销
	费用报告ID ExpenseReportId	每份费用报告的唯一标识，是该流程的主要案例标识。
说明 Expense Report ID会将单次费用提交相关的所有事件和活动归为一组，从费用最初录入到最终付款，完整、按时间顺序跟踪费用申报过程。在流程挖掘中，该属性对于重建每份费用报告的端到端历程至关重要。将其用作案例ID后，分析可以准确计算周期时间、识别瓶颈，并可视化报告在审批流程中经过的不同路径。为什么重要这是将所有相关活动关联到单个流程实例的基础属性，使端到端分析成为可能。获取位置该标识应位于Ramp中的主费用报告表或交易表内。示例 ER-2023-08-1123ER-2023-09-4591ER-2023-10-0024
	修订原因 RevisionReason	费用报告退回员工修改时提供的原因。
说明审批人拒绝或退回费用报告时，通常需要提供原因。此属性记录具体原因，例如“缺少收据”“类别错误”或“不符合政策”。这些信息对“费用修订率及原因”仪表板非常有价值。通过分析最常见的返工原因，组织可以发现提交流程中的系统性问题，并实施针对性的培训或系统改进，减少错误。为什么重要它直接揭示返工的根本原因，支持采取针对性措施，提高首次提交质量。获取位置当Ramp中发生“费用退回修改”活动时，这些数据会记录在评论或拒绝详情中。示例缺少分项收据费用超过政策限额选择了错误的费用类别重复交易
	员工部门 EmployeeDepartment	提交费用报告的员工所属部门。
说明该属性表示提交费用的员工所属业务单元或部门，例如“Sales”“Engineering”或“Marketing”。这是分析中的关键维度，可用于筛选和比较组织不同部门的流程绩效。它能够揭示各部门在审批时间、退回修改率和政策合规性方面的差异，为有针对性的改进提供支持。为什么重要它支持跨不同业务单元比较流程指标，突出效率、合规性和支出方面的差异。获取位置该信息可能关联于Ramp中的员工用户档案，或集成的人力资源系统。示例销售市场营销工程财务
	报告总金额 ReportTotalAmount	费用报告的货币总金额。
说明此属性表示单份报告中所有费用的总和，是了解支出模式的重要财务指标。在流程分析中，报告金额可用于划分案例，并调查高价值报告是否采用不同的审批路径，或需要更长处理时间。它是支出分析仪表板和发现节省成本机会的基础。为什么重要它为分析提供关键的财务维度，支持按金额划分报告并跟踪总体支出。获取位置这是Ramp费用报告对象中的主要字段。示例 150.752500.0089.99
	政策违规标记 PolicyViolationFlag	用于表示费用报告是否因违反政策而被标记。
说明如果系统的自动检查发现可能违反公司费用政策的情况，例如超出支出限额或提交重复费用，此布尔属性将设为true。此标记对“政策违规检测”仪表板及相关KPI至关重要。它有助于衡量政策控制的有效性，发现常见的不合规领域，为政策更新或员工培训提供依据。为什么重要它直接衡量政策合规情况，帮助发现并减少不合规支出及相关风险。获取位置这可能是Ramp中的系统生成标记，由提交或审批过程中的自动政策检查触发。示例 truefalse
	用户名 UserName	执行活动的用户姓名或ID，包括提交费用的员工或审批人。
说明该属性标识流程中某个事件的负责人，例如提交、批准或审核费用报告的人员。它可以是员工姓名或唯一用户ID。按用户分析有助于了解工作量分配、识别高绩效人员，并定位可能需要额外培训的人员。它也是审批绩效和资源管理相关仪表板的关键属性。为什么重要它将流程活动归属到具体人员，从而支持资源层面的绩效分析并识别培训需求。获取位置用户信息通常记录在Ramp中每份费用报告的审计轨迹或交易历史里。示例 Alice JohnsonBob SmithCharlie Brown系统自动化
	费用类别 ExpenseCategory	为费用分配的类别，例如“差旅”“软件”或“餐饮”。
说明此属性将费用归入预定义类别，有助于跟踪和控制公司支出。一份费用报告可能包含多个类别的项目。按费用类别分析是“支出类别分析”仪表板的基础。它帮助财务团队了解资金流向、监控预算执行情况，并发现支出随时间变化的趋势或异常。为什么重要它支持详细的支出分析，帮助发现主要成本驱动因素和优化预算的机会。获取位置这通常是Ramp费用报告中的明细行级数据。对于某些分析，可能需要将数据汇总到报告级别。示例机票餐饮和娱乐软件订阅办公用品
	事件结束时间 EventEndTime	表示具有持续时间的活动结束时间的时间戳。
说明许多活动是瞬时完成的，但“已执行政策检查”等活动可能具有可测量的持续时间。此属性记录这类活动的结束时间，与StartTime配合使用。同时记录开始和结束时间，才能准确计算活动处理时长。这对于“平均政策检查时长”等KPI至关重要，也有助于准确了解整体流程中具体自动化或人工任务的完成时间。为什么重要它支持准确计算各项活动的持续时间，这对于发现低效流程步骤至关重要。获取位置对于具有持续时间的活动，该值会记录在Ramp的事件数据中。对于瞬时事件，它可以与StartTime相同。示例 2023-10-26T10:00:05Z2023-10-26T14:35:10Z2023-10-27T09:10:00Z
	最近数据更新时间 LastDataUpdate	表示数据最近一次从源系统刷新的时间戳。
说明该属性记录最近一次数据提取的日期和时间，说明当前分析数据的新鲜度。在任何分析中，了解数据的时效性对于正确解读结果都至关重要。该属性可帮助用户判断所查看的信息是否为最新。为什么重要它告知用户数据的时效性，确保分析基于当前且相关的信息。获取位置该时间戳在数据提取过程中生成并添加。示例 2023-11-01T06:00:00Z
	审批步骤数 ApprovalStepCount	费用报告经过的正式审批步骤总数。
说明此计算指标统计单份费用报告中发生的不同审批活动数量，例如“经理已批准”和“财务已批准”，用于量化审批工作流的复杂度。此属性用于“每份报告的平均审批步骤数”KPI和“简单费用审批路径”仪表板。结合报告金额或类别分析该数量，组织可以发现简单、低金额费用是否经历了过于复杂的审批流程。为什么重要它量化工作流复杂度，帮助发现简化机会，尤其适用于低风险报告。获取位置该值通过统计事件日志中每个案例内特定审批活动的发生次数计算得出。示例 123
	审批经理 ApprovingManager	执行审批步骤的经理姓名。
说明此属性用于识别负责审核和批准员工费用报告的经理，与提交报告的用户不同。跟踪审批经理是“经理审批周期时间”仪表板的基础。它支持分析审批工作量和绩效，突出审批速度较快的经理以及成为瓶颈的经理，从而帮助平衡工作量或提供额外支持。为什么重要它支持分析各审批人的绩效，帮助发现并解决审批工作流中的瓶颈。获取位置这部分信息属于Ramp中的审批工作流数据，在经理处理报告时记录。示例 Jane DoeJohn MillerSusan Chen
	审计结果 AuditOutcome	对费用报告进行内部或外部审计后得到的结果。
说明对于接受正式审计的费用报告，此属性记录最终结果，例如“批准”“部分拒付”或“需要补充信息”。这些数据是“费用审计绩效”仪表板的核心。它有助于评估审计流程的有效性，跟踪不同结果的发生频率，并了解审计发现带来的财务影响。为什么重要它衡量审计流程的有效性和结果，帮助发现合规情况及控制薄弱点。获取位置如果使用Ramp的审计模块或集成系统进行详细费用审计，结果会记录在那里。示例通过通过，但有备注已拒绝已升级处理
	报销方式 ReimbursementMethod	执行报销付款所采用的方式，例如ACH或电汇。
说明此属性指定员工获得报销所使用的付款渠道。不同方式的处理时间和相关成本可能不同。按报销方式分析绩效，有助于发现最高效的付款渠道。“报销方式绩效”仪表板利用这些数据比较周期时间和可靠性，为优化付款策略提供依据。为什么重要它支持比较不同付款渠道的绩效，帮助优化付款速度和可靠性。获取位置这部分信息应记录在Ramp的付款或报销记录中。示例 ACH转账公司卡入账直接存款
	提交方式 SubmissionMethod	提交费用报告所使用的渠道，例如移动应用或Web门户。
说明此属性表示员工提交费用报告的方式。常见方式包括使用移动应用、桌面浏览器或通过邮件转发。分析提交方式可以了解用户行为和技术采用情况。例如，通过某一渠道提交的报告修订率较高，可能说明该渠道的界面存在易用性问题。为什么重要它提供用户行为背景，帮助判断某些提交渠道是否与较高错误率或延迟相关。获取位置这部分信息可能记录在Ramp系统日志中提交事件的元数据里。示例移动应用Web门户电子邮件
	是否返工 IsRework	用于表示费用报告是否至少被退回修改过一次的计算标记。
说明此布尔属性通过检查指定案例中是否发生“费用退回修改”活动得出。凡经历过至少一轮修订的报告，该值均设为true。此标记简化了“费用报告修订率”KPI的计算，并支持轻松筛选和比较首次通过审批的报告与需要返工的报告。分析这两个群组，可以揭示修订带来的时间和成本影响。为什么重要它可以快速划分流程，支持返工分析，帮助量化报告被退回修改的频率及影响。获取位置这是一个计算字段，在数据转换过程中通过检查案例内是否存在修订活动得出。示例 truefalse
	源系统 SourceSystem	标识数据提取来源的应用程序。
说明该属性指定流程数据的记录系统，本例中为Ramp。在数据可能来自多个系统的环境中，它有助于明确数据血缘。在分析中，它可用于筛选特定来源的数据，也可用于数据验证和治理。为什么重要它提供数据来源的背景信息，对于数据治理以及整合多个系统的数据至关重要。获取位置通常是在数据提取和转换过程中添加的静态值（“Ramp”）。示例 Ramp
	财务审批人 FinanceApprover	批准费用报告的财务团队用户。
说明对于包含财务审核步骤的工作流，此属性用于识别财务部门中执行最终审批的具体人员或团队。此属性支持“财务审核效率”仪表板，便于按个人或团队分析绩效。它有助于发现瓶颈、评估工作量分配，并衡量财务审核阶段的有效性。为什么重要它支持对财务审核步骤进行详细绩效分析，帮助优化流程中的关键控制点。获取位置这部分信息会在Ramp费用报告的审批历史中，于财务审核阶段记录。示例财务团队ADavid Lee财务自动化机器人

必需建议可选

费用管理活动

以下是应在事件日志中记录的关键流程步骤和里程碑，用于准确发现流程并衡量绩效。

5 建议 7 可选

	活动	说明
	产生费用	表示费用创建的时刻。通常，当员工使用公司卡，或手动创建自付费用记录时，系统会自动触发该事件。该事件一般来自交易数据流或用户界面操作。
为什么重要这是费用生命周期的主要开始事件。分析从该事件开始经过的时间，有助于了解提交延迟和整体流程速度。获取位置该事件来自Ramp卡交易日志，或手动录入费用对象的创建时间戳。请在费用表或交易表中查找初始记录创建事件。采集卡交易完成处理，或用户创建新的费用条目时，系统会直接记录该事件。事件类型 explicit
	已与会计系统同步	费用交易数据已成功写入集成的会计系统，例如NetSuite、QuickBooks或Xero。该事件标志着流程中的财务记录处理完成。
为什么重要这是端到端流程中的最后一个活动。此处延迟会影响财务报告的准确性和财务结账速度。获取位置该事件记录在集成日志中，或作为Ramp中费用对象的状态更新。请查找“Synced”“Posted”或“Exported”等状态。采集会计集成服务成功同步数据后，会创建一条日志记录。事件类型 explicit
	已执行报销	报销款已成功处理并支付给员工。这通常是员工侧的最后一步，也标志着付款周期结束。
为什么重要这是报销流程的主要结束事件。从提交到该事件的持续时间，是衡量员工满意度和流程效率的关键KPI。获取位置该事件来自付款处理日志或与支付服务商的集成。Ramp中的费用状态会更新为“Reimbursed”或“Paid”。采集付款系统确认转账成功后，系统会记录该事件。事件类型 explicit
	已提交费用	员工确认费用所需信息已完整，并将其提交至审批流程。这是一次明确的用户操作，会将费用从“草稿”或“需要处理”状态转为“待审批”状态。
为什么重要该活动是一个关键里程碑，标志着审批和报销周期正式开始，也是衡量审批及报销SLA的基准。获取位置该事件来自费用对象的状态历史，对应用户提交交易以供审核的操作。采集用户点击“Submit”按钮并触发状态变更时，系统会记录该事件。事件类型 explicit
	经理已批准	经理已审核并批准该费用，费用可进入下一步，例如财务审核或报销。该事件通过明确的用户操作记录。
为什么重要这是完成第一层审批的关键里程碑，对于分析审批工作流和识别瓶颈至关重要。获取位置经理点击“Approve”后，系统会在费用审批历史中记录该事件。事件日志应包含审批人的ID和时间戳。采集拥有经理权限的用户执行“Approve”操作后，系统会记录该事件。事件类型 explicit
	已安排报销	对于自付费用，已批准金额进入付款处理队列。该事件表示费用已通过所有审批，准备付款。
为什么重要该里程碑将审批流程与付款执行流程分开，有助于区分付款处理延迟和审批延迟。获取位置最终批准后，费用状态变为“Pending Reimbursement”或“Ready for Payment”时，通常可推断出该事件。如果付款采用批处理，也可能存在明确事件。采集根据状态变为“Ready for Payout”，或付款批次表中创建记录的时间确定。事件类型 inferred
	已执行政策检查	系统会根据已配置的公司政策自动审核费用，并标记潜在违规。这通常是提交后不久发生的系统生成事件。
为什么重要衡量自动化一致性检查的效率及其对流程的影响。帮助识别常见的政策违规和员工培训领域。获取位置该事件可能记录在与费用交易关联的审计轨迹或日志中。请查找与“policy_check”或“compliance_scan”相关的系统事件。采集自动政策引擎完成交易检查后，系统会创建一条日志记录。事件类型 explicit
	已附加收据	表示收据与费用关联的时刻，可以通过OCR匹配自动完成，也可以由用户手动完成。当收据文件成功关联到交易记录时，系统会记录该事件。
为什么重要跟踪此活动有助于识别因缺少凭证造成的延迟，也是确保合规和做好审计准备的关键步骤。获取位置收据上传或匹配后，系统会在费用或交易历史中记录该事件。请查找附件创建事件，或表示“receipt_attached”的标记。采集系统将收据图像或文件关联到费用记录时，会创建该事件。事件类型 explicit
	待经理审核	费用已提交，正在等待员工直属经理审核。当费用提交后，状态变为“Pending Manager Approval”或类似值时，可推断出该状态。
为什么重要标志着经理审批阶段开始。分析费用在此状态中停留的时间，是衡量和改进经理审批周期的关键。获取位置当费用对象状态变为“Pending Approval”等状态，并被分配到经理队列时，可推断出该事件。需要跟踪状态历史。采集根据费用状态变为“Pending Manager Approval”时的时间戳确定。事件类型 inferred
	待财务审核	费用获批后已升级处理，目前正在等待财务或会计团队审核。高金额费用或存在政策标记的费用通常会进入此阶段。该活动可根据状态变更推断。
为什么重要标志着财务审核周期开始。衡量该阶段的持续时间，有助于评估财务团队的工作量和效率，并识别自动化机会。获取位置经理批准后，费用对象状态变为“Pending Finance Approval”时，可推断出该事件。需要访问费用状态历史。采集根据费用状态更新为“Pending Finance Review”时的时间戳确定。事件类型 inferred
	财务已批准	财务团队已完成审核并最终批准费用，费用可进入报销和会计同步环节。该事件由财务团队成员执行的明确用户操作记录。
为什么重要表示付款前的最终审批关口。分析该活动有助于了解端到端审批周期和财务团队效率。获取位置该事件会记录在费用审批历史中。请查找与财务部门用户关联的审批事件。采集拥有财务权限的用户执行“Approve”操作后，系统会记录该事件。事件类型 explicit
	费用退回修改	经理或财务审核人员拒绝了该费用，并将其退回员工修改。系统会将状态变更为“Needs Revision”或“Rejected”等状态。
为什么重要该活动表示流程中的返工循环，会直接增加周期时间。跟踪这些事件有助于识别常见提交错误，并提升一次通过率。获取位置当费用对象状态变为“Needs Revision”或类似状态时，可推断出该事件。事件应关联发起该操作的审批人。采集根据费用状态变为“Needs Revision”或“Rejected”时的时间戳确定。事件类型 inferred

建议可选

提取指南

如何从Ramp获取数据

步骤

使用Admin或Bookkeeper权限登录您的Ramp账户。
从左侧菜单进入交易部分。此视图包含所有企业卡支出。
对数据应用所需筛选条件。至少设置要分析期间的Date Range。如有需要，您还可以按Status、Category或Department等其他属性筛选，以缩小范围。
点击交易列表右上方的Export按钮。
在导出窗口中，选择交易作为导出类型，选择CSV作为格式。下载文件并将其命名为transactions_export.csv。
接下来，从左侧菜单进入报销部分。此视图包含员工提交的所有个人垫付费用。
应用与交易导出相同的日期范围筛选条件，确保数据一致。
点击Export按钮，将报销数据下载为CSV文件。将文件命名为reimbursements_export.csv。
将导出的两个CSV文件（transactions_export.csv和reimbursements_export.csv）放在提供的Python脚本所在目录中。
执行Python脚本。脚本会加载两个文件，为每个业务事件创建单独的行，转换数据，并将其合并为一个结构化事件日志。
脚本将输出名为ramp_event_log.csv的文件。该文件符合ProcessMind的要求。
将ramp_event_log.csv文件上传到ProcessMind进行分析。

配置

权限：您必须在Ramp中拥有“Admin”或“Bookkeeper”角色，才能访问并导出完整的交易和报销数据集。
日期范围：首次分析建议选择至少6至12个月的日期范围，以捕获足够的流程变化。持续监控时，可以使用更短且更近期的范围。
数据一致性：导出Transactions和Reimbursements文件时，确保使用完全相同的日期范围和筛选条件，以保持数据集一致。
导出格式：两个文件始终选择CSV格式，因为转换脚本用于处理逗号分隔文件。
文件命名：使用transactions_export.csv和reimbursements_export.csv作为文件名；如需使用其他名称，请相应更新转换脚本中的文件路径变量。

a 示例查询 python

import pandas as pd
import numpy as np

class=class="hl-string">"hl-comment"># --- Configuration ---
TRANSACTIONS_FILE = class="hl-string">'transactions_export.csv'
REIMBURSEMENTS_FILE = class="hl-string">'reimbursements_export.csv'
OUTPUT_FILE = class="hl-string">'ramp_event_log.csv'

class=class="hl-string">"hl-comment"># --- Load Data ---
def load_data():
    try:
        transactions_df = pd.read_csv(TRANSACTIONS_FILE)
        reimbursements_df = pd.read_csv(REIMBURSEMENTS_FILE)
        print(class="hl-string">"Successfully loaded transactions and reimbursements files.")
        return transactions_df, reimbursements_df
    except FileNotFoundError as e:
        print(fclass="hl-string">"Error: {e}. Make sure '{TRANSACTIONS_FILE}' and '{REIMBURSEMENTS_FILE}' are in the same directory.")
        return None, None

class=class="hl-string">"hl-comment"># --- Main Transformation Logic ---
def transform_to_event_log(transactions_df, reimbursements_df):
    events = []

    class=class="hl-string">"hl-comment"># Standardize column names for easier processing
    class=class="hl-string">"hl-comment"># These names are based on typical Ramp CSV exports, adjust if your export has different headers
    transactions_df.rename(columns={
        class="hl-string">'Transaction ID': class="hl-string">'ExpenseReportId',
        class="hl-string">'User Name': class="hl-string">'UserName',
        class="hl-string">'Department': class="hl-string">'EmployeeDepartment',
        class="hl-string">'Amount': class="hl-string">'ReportTotalAmount',
        class="hl-string">'Category Name': class="hl-string">'ExpenseCategory',
        class="hl-string">'Transaction Date': class="hl-string">'ExpenseIncurredTimestamp',
        class="hl-string">'Receipts Attached At': class="hl-string">'ReceiptAttachedTimestamp',
        class="hl-string">'Submitted At': class="hl-string">'ExpenseSubmittedTimestamp',
        class="hl-string">'Manager Approved At': class="hl-string">'ManagerApprovedTimestamp',
        class="hl-string">'Needs Review At': class="hl-string">'FinanceReviewPendingTimestamp',
        class="hl-string">'Admin Approved At': class="hl-string">'FinanceApprovedTimestamp',
        class="hl-string">'Accounting Sync At': class="hl-string">'SyncedTimestamp',
        class="hl-string">'Memo': class="hl-string">'RevisionReason',
        class="hl-string">'State': class="hl-string">'Status'
    }, inplace=True)

    reimbursements_df.rename(columns={
        class="hl-string">'Reimbursement ID': class="hl-string">'ExpenseReportId',
        class="hl-string">'User Name': class="hl-string">'UserName',
        class="hl-string">'Department': class="hl-string">'EmployeeDepartment',
        class="hl-string">'Amount': class="hl-string">'ReportTotalAmount',
        class="hl-string">'Category Name': class="hl-string">'ExpenseCategory',
        class="hl-string">'Created At': class="hl-string">'ReimbursementScheduledTimestamp',
        class="hl-string">'Paid Out At': class="hl-string">'ReimbursementExecutedTimestamp',
        class="hl-string">'Memo': class="hl-string">'RevisionReason',
        class="hl-string">'State': class="hl-string">'Status'
    }, inplace=True)

    class=class="hl-string">"hl-comment"># Combine card transactions and reimbursements into a single dataframe for processing common events
    class=class="hl-string">"hl-comment"># Note: Reimbursements will be processed separately for reimbursement-specific events
    combined_df = transactions_df

    class=class="hl-string">"hl-comment"># Process activities from the combined dataframe
    for _, row in combined_df.iterrows():
        base_event = {
            class="hl-string">'ExpenseReportId': row.get(class="hl-string">'ExpenseReportId'),
            class="hl-string">'UserName': row.get(class="hl-string">'UserName'),
            class="hl-string">'EmployeeDepartment': row.get(class="hl-string">'EmployeeDepartment'),
            class="hl-string">'ReportTotalAmount': row.get(class="hl-string">'ReportTotalAmount'),
            class="hl-string">'ExpenseCategory': row.get(class="hl-string">'ExpenseCategory'),
            class="hl-string">'PolicyViolationFlag': class="hl-string">'FLAGGED' in str(row.get(class="hl-string">'Flags')), class=class="hl-string">"hl-comment"># Infer from class="hl-string">'Flags' column
            class="hl-string">'RevisionReason': row.get(class="hl-string">'RevisionReason')
        }

        class=class="hl-string">"hl-comment"># Activity: Expense Incurred
        if pd.notna(row.get(class="hl-string">'ExpenseIncurredTimestamp')):
            events.append({**base_event, class="hl-string">'ActivityName': class="hl-string">'Expense Incurred', class="hl-string">'EventTimestamp': row.get(class="hl-string">'ExpenseIncurredTimestamp')})

        class=class="hl-string">"hl-comment"># Activity: Receipt Attached
        if pd.notna(row.get(class="hl-string">'ReceiptAttachedTimestamp')):
            events.append({**base_event, class="hl-string">'ActivityName': class="hl-string">'Receipt Attached', class="hl-string">'EventTimestamp': row.get(class="hl-string">'ReceiptAttachedTimestamp')})

        class=class="hl-string">"hl-comment"># Activity: Expense Submitted
        if pd.notna(row.get(class="hl-string">'ExpenseSubmittedTimestamp')):
            events.append({**base_event, class="hl-string">'ActivityName': class="hl-string">'Expense Submitted', class="hl-string">'EventTimestamp': row.get(class="hl-string">'ExpenseSubmittedTimestamp')})
            class=class="hl-string">"hl-comment"># Inferred Activity: Policy Check Performed (assumed to happen right after submission)
            policy_check_ts = pd.to_datetime(row.get(class="hl-string">'ExpenseSubmittedTimestamp')) + pd.Timedelta(seconds=1)
            events.append({**base_event, class="hl-string">'ActivityName': class="hl-string">'Policy Check Performed', class="hl-string">'EventTimestamp': policy_check_ts.isoformat()})
            class=class="hl-string">"hl-comment"># Inferred Activity: Manager Review Pending (starts when submitted)
            events.append({**base_event, class="hl-string">'ActivityName': class="hl-string">'Manager Review Pending', class="hl-string">'EventTimestamp': row.get(class="hl-string">'ExpenseSubmittedTimestamp')})

        class=class="hl-string">"hl-comment"># Activity: Manager Approved
        if pd.notna(row.get(class="hl-string">'ManagerApprovedTimestamp')):
            events.append({**base_event, class="hl-string">'ActivityName': class="hl-string">'Manager Approved', class="hl-string">'EventTimestamp': row.get(class="hl-string">'ManagerApprovedTimestamp')})

        class=class="hl-string">"hl-comment"># Activity: Finance Review Pending
        if pd.notna(row.get(class="hl-string">'FinanceReviewPendingTimestamp')):
            events.append({**base_event, class="hl-string">'ActivityName': class="hl-string">'Finance Review Pending', class="hl-string">'EventTimestamp': row.get(class="hl-string">'FinanceReviewPendingTimestamp')})

        class=class="hl-string">"hl-comment"># Activity: Finance Approved
        if pd.notna(row.get(class="hl-string">'FinanceApprovedTimestamp')):
            events.append({**base_event, class="hl-string">'ActivityName': class="hl-string">'Finance Approved', class="hl-string">'EventTimestamp': row.get(class="hl-string">'FinanceApprovedTimestamp')})

        class=class="hl-string">"hl-comment"># Activity: Synced with Accounting System
        if pd.notna(row.get(class="hl-string">'SyncedTimestamp')):
            events.append({**base_event, class="hl-string">'ActivityName': class="hl-string">'Synced with Accounting System', class="hl-string">'EventTimestamp': row.get(class="hl-string">'SyncedTimestamp')})

        class=class="hl-string">"hl-comment"># Activity: Expense Sent Back for Revision
        if row.get(class="hl-string">'Status') == class="hl-string">'REJECTED' and pd.notna(row.get(class="hl-string">'Updated At')):
             events.append({**base_event, class="hl-string">'ActivityName': class="hl-string">'Expense Sent Back for Revision', class="hl-string">'EventTimestamp': row.get(class="hl-string">'Updated At')}) class=class="hl-string">"hl-comment"># Using last updated time for rejection

    class=class="hl-string">"hl-comment"># Process reimbursement-specific activities
    for _, row in reimbursements_df.iterrows():
        base_event = {
            class="hl-string">'ExpenseReportId': row.get(class="hl-string">'ExpenseReportId'),
            class="hl-string">'UserName': row.get(class="hl-string">'UserName'),
            class="hl-string">'EmployeeDepartment': row.get(class="hl-string">'EmployeeDepartment'),
            class="hl-string">'ReportTotalAmount': row.get(class="hl-string">'ReportTotalAmount'),
            class="hl-string">'ExpenseCategory': row.get(class="hl-string">'ExpenseCategory'),
            class="hl-string">'PolicyViolationFlag': False, class=class="hl-string">"hl-comment"># Typically less flagging on reimbursements
            class="hl-string">'RevisionReason': row.get(class="hl-string">'RevisionReason')
        }

        class=class="hl-string">"hl-comment"># Activity: Reimbursement Scheduled
        if row.get(class="hl-string">'Status') == class="hl-string">'APPROVED' and pd.notna(row.get(class="hl-string">'ReimbursementScheduledTimestamp')):
            events.append({**base_event, class="hl-string">'ActivityName': class="hl-string">'Reimbursement Scheduled', class="hl-string">'EventTimestamp': row.get(class="hl-string">'ReimbursementScheduledTimestamp')})

        class=class="hl-string">"hl-comment"># Activity: Reimbursement Executed
        if row.get(class="hl-string">'Status') == class="hl-string">'PAID' and pd.notna(row.get(class="hl-string">'ReimbursementExecutedTimestamp')):
            events.append({**base_event, class="hl-string">'ActivityName': class="hl-string">'Reimbursement Executed', class="hl-string">'EventTimestamp': row.get(class="hl-string">'ReimbursementExecutedTimestamp')})

    if not events:
        print(class="hl-string">"Warning: No events were generated. Check the input files and column names.")
        return None

    class=class="hl-string">"hl-comment"># Create final DataFrame
    event_log_df = pd.DataFrame(events)
    event_log_df[class="hl-string">'EventTimestamp'] = pd.to_datetime(event_log_df[class="hl-string">'EventTimestamp'])
    event_log_df.sort_values(by=[class="hl-string">'ExpenseReportId', class="hl-string">'EventTimestamp'], inplace=True)
    
    return event_log_df

class=class="hl-string">"hl-comment"># --- Main execution block ---
if __name__ == class="hl-string">"__main__":
    transactions_df, reimbursements_df = load_data()
    if transactions_df is not None and reimbursements_df is not None:
        event_log = transform_to_event_log(transactions_df, reimbursements_df)
        if event_log is not None:
            event_log.to_csv(OUTPUT_FILE, index=False, date_format=class="hl-string">'%Y-%m-%dT%H:%M:%S.%fZ')
            print(fclass="hl-string">"Successfully created event log with {len(event_log)} events. Saved to '{OUTPUT_FILE}'.")

步骤

前提条件和环境设置：确保已安装Python3.8或更高版本。运行pip install requests pandas安装所需Python库。您还必须拥有Ramp实例的管理员权限，以生成API凭据。
生成Ramp API凭据：登录Ramp账户。进入Settings，然后从侧边菜单选择Developer API。点击“Create App”，为应用命名并配置所需权限范围。您需要transactions:read、reimbursements:read和users:read。创建完成后，安全地复制Client ID和Client Secret。
配置Python脚本：新建Python文件，例如extract_ramp_log.py。将提供的Python脚本复制到该文件中。将Client ID和Client Secret设置为环境变量，或为测试目的直接写入脚本。出于安全考虑，强烈建议使用环境变量。
设置提取日期范围：在脚本的main执行代码块中找到START_DATE和END_DATE变量。调整这些日期以定义数据提取时间范围。初次分析建议使用3至6个月的时间段。
执行脚本：在终端中运行命令python extract_ramp_log.py。脚本将通过Ramp API完成身份验证，从users、transactions和reimbursements端点获取数据，并自动处理分页。
了解API调用：脚本首先获取所有用户数据，建立用户ID与姓名、部门之间的映射。随后获取指定日期范围内的所有报销记录，并提取每笔报销的审批和付款历史。
将数据转换为事件日志：脚本核心逻辑会遍历每笔报销及其关联状态历史，将“MANAGER_APPROVAL”或“FINISHED”等状态变更转换为“Manager Review Pending”或“Reimbursement Executed”等独立活动。同时，脚本还会获取关联交易数据，创建“Expense Incurred”和“Receipt Attached”事件。
处理数据丰富：脚本通过为每个事件添加用户详情、部门信息和费用类别来丰富事件日志，从而生成用于流程分析的完整数据集。
检查输出：脚本完成后，会在同一目录中生成名为ramp_event_log.csv的文件。请使用电子表格程序打开该CSV文件，检查数据并确认提取结果符合预期。
准备上传：生成的CSV文件包含必需列ExpenseReportId、ActivityName、EventTimestamp以及其他推荐属性，已可直接上传到ProcessMind平台。

配置

API凭据：身份验证需要有效的Client ID和Client Secret。这些凭据可在Settings > Developer API下的Ramp仪表板中生成。
API权限范围：脚本需要特定权限才能访问数据。请确保API密钥已启用以下权限范围：transactions:read（费用和收据详情）、reimbursements:read（提交、审批和付款状态）以及users:read（员工姓名和部门）。
日期范围：提取范围由脚本中的START_DATE和END_DATE参数控制。建议先选择近期时间段，例如最近3至6个月，以控制初始数据集规模。筛选依据为报销创建日期。
分页：脚本使用Ramp API提供的starting_after游标自动处理分页。每次请求最多获取100条记录，确保提取日期范围内的全部数据。
速率限制：Ramp API通常限制为每分钟600次请求。所提供脚本通常运行高效，但对于超大数据集，如果遇到速率限制错误，可能需要在API调用之间加入短暂延迟。

a 示例查询 python

import os
import requests
import pandas as pd
from datetime import datetime, timezone

class=class="hl-string">"hl-comment"># --- Configuration ---
class=class="hl-string">"hl-comment"># It is recommended to use environment variables for credentials
CLIENT_ID = os.getenv(class="hl-string">'RAMP_CLIENT_ID', class="hl-string">'[Your_Ramp_Client_ID]')
CLIENT_SECRET = os.getenv(class="hl-string">'RAMP_CLIENT_SECRET', class="hl-string">'[Your_Ramp_Client_Secret]')

API_BASE_URL = class="hl-string">'https://api.ramp.com/v1'
TOKEN_URL = class="hl-string">'https://api.ramp.com/v1/public/customer/token'

class=class="hl-string">"hl-comment"># --- Main Functions ---

def get_access_token():
    class="hl-string">"""Authenticates with the Ramp API to get a bearer token."""
    headers = {class="hl-string">'Content-Type': class="hl-string">'application/x-www-form-urlencoded'}
    data = {
        class="hl-string">'grant_type': class="hl-string">'client_credentials',
        class="hl-string">'client_id': CLIENT_ID,
        class="hl-string">'client_secret': CLIENT_SECRET,
        class="hl-string">'scope': class="hl-string">'transactions:read reimbursements:read users:read'
    }
    try:
        response = requests.post(TOKEN_URL, headers=headers, data=data)
        response.raise_for_status()
        return response.json()[class="hl-string">'access_token']
    except requests.exceptions.RequestException as e:
        print(fclass="hl-string">"Error obtaining access token: {e}")
        if e.response is not None:
            print(fclass="hl-string">"Response body: {e.response.text}")
        return None

def fetch_paginated_data(endpoint, access_token, params=None):
    class="hl-string">"""Fetches all data from a paginated Ramp API endpoint."""
    all_data = []
    headers = {class="hl-string">'Authorization': fclass="hl-string">'Bearer {access_token}'}
    url = fclass="hl-string">"{API_BASE_URL}{endpoint}"
    
    while url:
        try:
            response = requests.get(url, headers=headers, params=params)
            response.raise_for_status()
            data = response.json()
            all_data.extend(data.get(class="hl-string">'data', []))
            url = data.get(class="hl-string">'next') class=class="hl-string">"hl-comment"># Ramp uses a full URL for the next page
            params = None class=class="hl-string">"hl-comment"># Subsequent requests use the class="hl-string">'next' URL which includes params
        except requests.exceptions.RequestException as e:
            print(fclass="hl-string">"Error fetching data from {endpoint}: {e}")
            break
            
    return all_data

def map_state_to_activity(state, reason_code):
    class="hl-string">"""Maps reimbursement state history to a user-friendly activity name."""
    if reason_code == class="hl-string">'REJECTED':
        return class="hl-string">'Expense Sent Back for Revision'
    
    state_map = {
        class="hl-string">'SUBMITTED': class="hl-string">'Expense Submitted',
        class="hl-string">'MANAGER_APPROVAL': class="hl-string">'Manager Review Pending',
        class="hl-string">'ACCOUNTING_APPROVAL': class="hl-string">'Finance Review Pending',
        class="hl-string">'MANAGER_APPROVED': class="hl-string">'Manager Approved',
        class="hl-string">'ACCOUNTING_APPROVED': class="hl-string">'Finance Approved',
        class="hl-string">'PAYMENT_SCHEDULED': class="hl-string">'Reimbursement Scheduled',
        class="hl-string">'PAYMENT_PROCESSING': class="hl-string">'Reimbursement Executed', class=class="hl-string">"hl-comment"># Approximation
        class="hl-string">'FINISHED': class="hl-string">'Reimbursement Executed',
        class="hl-string">'SYNCED': class="hl-string">'Synced with Accounting System'
    }
    return state_map.get(state)

class=class="hl-string">"hl-comment"># --- Main Execution ---
def main():
    class="hl-string">"""Main script to extract, transform, and save the event log."""
    print(class="hl-string">"Starting Ramp data extraction...")
    access_token = get_access_token()
    if not access_token:
        print(class="hl-string">"Failed to get access token. Exiting.")
        return

    class=class="hl-string">"hl-comment"># 1. Fetch Users for enrichment
    print(class="hl-string">"Fetching users...")
    users = fetch_paginated_data(class="hl-string">'/users', access_token)
    user_map = {u[class="hl-string">'id']: {class="hl-string">'name': fclass="hl-string">"{u['first_name']} {u['last_name']}", class="hl-string">'department': u.get(class="hl-string">'department')} for u in users}
    print(fclass="hl-string">"Found {len(user_map)} users.")

    class=class="hl-string">"hl-comment"># 2. Fetch Reimbursements within a date range
    START_DATE = class="hl-string">'2023-01-01T00:00:00Z'
    END_DATE = class="hl-string">'2023-12-31T23:59:59Z'
    print(fclass="hl-string">"Fetching reimbursements from {START_DATE} to {END_DATE}...")
    reimbursement_params = {class="hl-string">'created_at_gte': START_DATE, class="hl-string">'created_at_lte': END_DATE}
    reimbursements = fetch_paginated_data(class="hl-string">'/reimbursements', access_token, params=reimbursement_params)
    print(fclass="hl-string">"Found {len(reimbursements)} reimbursements.")

    event_log = []

    class=class="hl-string">"hl-comment"># 3. Process each reimbursement to create the event log
    for r in reimbursements:
        case_id = r[class="hl-string">'id']
        report_total = r[class="hl-string">'amount']
        submitter_id = r[class="hl-string">'user_id']
        submitter_details = user_map.get(submitter_id, {class="hl-string">'name': class="hl-string">'Unknown User', class="hl-string">'department': None})
        expense_category = r.get(class="hl-string">'merchant_name') class=class="hl-string">"hl-comment"># Best available proxy for category
        
        class=class="hl-string">"hl-comment"># Process reimbursement state history
        for history in r.get(class="hl-string">'state_history', []):
            activity_name = map_state_to_activity(history[class="hl-string">'state'], history.get(class="hl-string">'reason_code'))
            if activity_name:
                event_log.append({
                    class="hl-string">'ExpenseReportId': case_id,
                    class="hl-string">'ActivityName': activity_name,
                    class="hl-string">'EventTimestamp': history[class="hl-string">'created_at'],
                    class="hl-string">'UserName': user_map.get(history.get(class="hl-string">'user_id'), submitter_details)[class="hl-string">'name'],
                    class="hl-string">'EmployeeDepartment': submitter_details[class="hl-string">'department'],
                    class="hl-string">'ReportTotalAmount': report_total,
                    class="hl-string">'ExpenseCategory': expense_category,
                    class="hl-string">'PolicyViolationFlag': None, class=class="hl-string">"hl-comment"># Not directly available in reimbursement history
                    class="hl-string">'RevisionReason': history.get(class="hl-string">'reason_notes')
                })

        class=class="hl-string">"hl-comment"># Fetch and process associated transactions
        for tx_id in r.get(class="hl-string">'transaction_ids', []):
            try:
                tx_response = requests.get(fclass="hl-string">"{API_BASE_URL}/transactions/{tx_id}", headers={class="hl-string">'Authorization': fclass="hl-string">'Bearer {access_token}'})
                tx_response.raise_for_status()
                tx = tx_response.json()

                class=class="hl-string">"hl-comment"># Activity: Expense Incurred
                event_log.append({
                    class="hl-string">'ExpenseReportId': case_id,
                    class="hl-string">'ActivityName': class="hl-string">'Expense Incurred',
                    class="hl-string">'EventTimestamp': tx[class="hl-string">'user_transaction_time'],
                    class="hl-string">'UserName': submitter_details[class="hl-string">'name'],
                    class="hl-string">'EmployeeDepartment': submitter_details[class="hl-string">'department'],
                    class="hl-string">'ReportTotalAmount': report_total,
                    class="hl-string">'ExpenseCategory': tx.get(class="hl-string">'merchant_name'),
                    class="hl-string">'PolicyViolationFlag': tx.get(class="hl-string">'flagged_for_review'),
                    class="hl-string">'RevisionReason': None
                })

                class=class="hl-string">"hl-comment"># Activity: Receipt Attached
                if tx.get(class="hl-string">'receipts') and len(tx[class="hl-string">'receipts']) > 0:
                    receipt_time = tx[class="hl-string">'receipts'][0].get(class="hl-string">'upload_time', tx[class="hl-string">'user_transaction_time']) class=class="hl-string">"hl-comment"># Use transaction time as fallback
                    event_log.append({
                        class="hl-string">'ExpenseReportId': case_id,
                        class="hl-string">'ActivityName': class="hl-string">'Receipt Attached',
                        class="hl-string">'EventTimestamp': receipt_time,
                        class="hl-string">'UserName': submitter_details[class="hl-string">'name'],
                        class="hl-string">'EmployeeDepartment': submitter_details[class="hl-string">'department'],
                        class="hl-string">'ReportTotalAmount': report_total,
                        class="hl-string">'ExpenseCategory': tx.get(class="hl-string">'merchant_name'),
                        class="hl-string">'PolicyViolationFlag': tx.get(class="hl-string">'flagged_for_review'),
                        class="hl-string">'RevisionReason': None
                    })

                class=class="hl-string">"hl-comment"># Activity: Policy Check Performed (Approximation)
                if tx.get(class="hl-string">'flagged_for_review'):
                    event_log.append({
                        class="hl-string">'ExpenseReportId': case_id,
                        class="hl-string">'ActivityName': class="hl-string">'Policy Check Performed',
                        class="hl-string">'EventTimestamp': r.get(class="hl-string">'created_at'), class=class="hl-string">"hl-comment"># Use submission time as proxy
                        class="hl-string">'UserName': class="hl-string">'System',
                        class="hl-string">'EmployeeDepartment': submitter_details[class="hl-string">'department'],
                        class="hl-string">'ReportTotalAmount': report_total,
                        class="hl-string">'ExpenseCategory': tx.get(class="hl-string">'merchant_name'),
                        class="hl-string">'PolicyViolationFlag': True,
                        class="hl-string">'RevisionReason': None
                    })

            except requests.exceptions.RequestException as e:
                print(fclass="hl-string">"Could not fetch transaction {tx_id}: {e}")

    if not event_log:
        print(class="hl-string">"No events were generated. Check the date range and API responses.")
        return

    class=class="hl-string">"hl-comment"># 4. Create and save DataFrame
    print(class="hl-string">"Finalizing event log...")
    df = pd.DataFrame(event_log)
    df[class="hl-string">'EventTimestamp'] = pd.to_datetime(df[class="hl-string">'EventTimestamp'])
    df = df.sort_values(by=[class="hl-string">'ExpenseReportId', class="hl-string">'EventTimestamp'])

    class=class="hl-string">"hl-comment"># Ensure required columns are present
    final_columns = [
        class="hl-string">'ExpenseReportId', class="hl-string">'ActivityName', class="hl-string">'EventTimestamp', class="hl-string">'UserName',
        class="hl-string">'EmployeeDepartment', class="hl-string">'ReportTotalAmount', class="hl-string">'ExpenseCategory',
        class="hl-string">'PolicyViolationFlag', class="hl-string">'RevisionReason'
    ]
    for col in final_columns:
        if col not in df.columns:
            df[col] = None
    
    df = df[final_columns]

    output_filename = class="hl-string">'ramp_event_log.csv'
    df.to_csv(output_filename, index=False, date_format=class="hl-string">'%Y-%m-%dT%H:%M:%S.%f%z')
    print(fclass="hl-string">"Extraction complete. Event log saved to {output_filename}")

if __name__ == class="hl-string">'__main__':
    main()

步骤

确认您的Ramp订阅包含原生数据仓库集成功能，并确认管理员可以配置目标位置。选择受支持的目标仓库，例如Snowflake或BigQuery，并按照组织的安全标准配置访问权限。具体菜单路径和目标对象取决于您的Ramp配置，请使用[Configure based on your system]。
配置仓库目标、数据库或项目、架构或数据集、区域、凭据以及同步范围。记录Ramp显示的复制表和列名。不要假设表名，因为它们会因集成方式和账户配置而异。
确定已复制的费用报告、费用交易、收据、审批、政策、报销和会计同步数据。将每个源映射到事件日志所需字段。如果某个必需生命周期事件未被明确记录，请确定记录该事件的源状态历史或审计数据。
使用[Start date]和[End date]定义提取期间，建议设置为三至六个月。仅在存在相应源字段时应用公司或组织筛选条件。保留源时区，或将所有时间戳转换为有文档说明的统一时区。
创建仓库视图或查询，将源字段标准化为ExpenseReportId、ActivityName、EventTimestamp、UserName、EmployeeDepartment、ReportTotalAmount、ExpenseCategory、PolicyViolationFlag和RevisionReason。将所提供查询中的每个方括号表名和列名占位符替换为复制架构中的实际名称。
将每个生命周期活动提取为明确的事件行。下面的查询为12个必需活动分别提供完整的SELECT代码块，不会从无关记录推断事件。对于基于状态的活动，请使用源状态历史或事件表，而不仅是当前报告状态，以保留历史变更。
验证标识符、时间戳、活动标签、重复记录处理方式和空值处理。确认每个事件都属于正确的ExpenseReportId；当源系统记录了多个合法发生的事件时，应保留重复活动。
将结果导出为CSV或其他ProcessMind支持的表格格式。使用准确的必需列名ExpenseReportId、ActivityName和EventTimestamp，并在可用时加入推荐属性，然后以Expense Report ID作为案例标识符上传文件。ProcessMind会按原样读取事件日志，因此不要期待平台自动生成缺失活动。

配置

集成目标：为Snowflake或BigQuery配置Ramp原生数据仓库集成，具体账户设置请使用[Configure based on your system]。
来源对象：使用数据仓库中提供的已复制费用报告、交易、收据、审批、政策、报销和会计同步对象。将[Your table name]和[Your column name]占位符替换为集成中已验证的名称。
日期范围：先使用三至六个月的数据。验证数据仓库成本、查询时长、事件量和ProcessMind上传限制后，再扩大时间范围。
日期筛选：如果目标是捕获期间内发生的全部活动，请按事件时间戳筛选，而不仅是报告创建时间戳。
组织筛选：仅当复制数据中存在Company Code、实体、部门、卡项目或其他相应字段时，才应用这些筛选条件。需要按公司限制时，请使用[Company filter]。
活动历史：优先使用不可变的审计记录或状态历史记录。仅依赖当前状态表，可能无法保留收据附加、提交、审批、修订、报销或会计同步历史。
时间戳处理：将时间戳统一到一个时区，并记录来源值是UTC还是本地时间。在可用时保留来源精度。
所需访问权限：管理员可能需要具备配置集成、创建或查询数据仓库对象，以及访问费用、审批、报销和会计数据的权限。
许可和可用性：数据仓库复制、历史审计数据、报销数据和会计同步数据是否可用，取决于Ramp套餐、已启用模块、集成配置和数据保留政策。
性能：尽早限制日期范围，仅选择必需列，在适当情况下按组织筛选；对于大型数据集，可创建中间标准化视图或表。根据目标数据仓库支持的功能进行分区或聚类。
增量运行：对于定期加载，使用已处理EventTimestamp的最大值作为水位线，并设置较小的重叠窗口。只有在确认来源事件标识符可靠时才进行去重，因为删除重复事件可能改变流程行为。

a 示例查询 python

import os
import pandas as pd
from sqlalchemy import create_engine, text

CONNECTION_STRING = os.getenv(class="hl-string">"WAREHOUSE_CONNECTION_STRING", class="hl-string">"[Connection string]")
START_DATE = class="hl-string">"[Start date]"
END_DATE = class="hl-string">"[End date]"
COMPANY_FILTER = class="hl-string">"[Company filter]"

engine = create_engine(CONNECTION_STRING)

sql = text(class="hl-string">"""
WITH
expense_incurred AS (
    SELECT
        [Expense report ID column] AS ExpenseReportId,
        'Expense Incurred' AS ActivityName,
        [Expense incurred timestamp column] AS EventTimestamp,
        [User name column] AS UserName,
        [Employee department column] AS EmployeeDepartment,
        [Report total amount column] AS ReportTotalAmount,
        [Expense category column] AS ExpenseCategory,
        [Policy violation flag column] AS PolicyViolationFlag,
        CAST(NULL AS VARCHAR(4000)) AS RevisionReason
    FROM [Your expense incurred table]
    WHERE [Expense incurred timestamp column] >= :start_date
      AND [Expense incurred timestamp column] < :end_date
      AND ([Company column] = :company_filter OR :company_filter = '[Company filter]')
),
receipt_attached AS (
    SELECT
        [Expense report ID column] AS ExpenseReportId,
        'Receipt Attached' AS ActivityName,
        [Receipt attached timestamp column] AS EventTimestamp,
        [User name column] AS UserName,
        [Employee department column] AS EmployeeDepartment,
        [Report total amount column] AS ReportTotalAmount,
        [Expense category column] AS ExpenseCategory,
        [Policy violation flag column] AS PolicyViolationFlag,
        CAST(NULL AS VARCHAR(4000)) AS RevisionReason
    FROM [Your receipt attachment table]
    WHERE [Receipt attached timestamp column] >= :start_date
      AND [Receipt attached timestamp column] < :end_date
      AND ([Company column] = :company_filter OR :company_filter = '[Company filter]')
),
expense_submitted AS (
    SELECT
        [Expense report ID column] AS ExpenseReportId,
        'Expense Submitted' AS ActivityName,
        [Submission timestamp column] AS EventTimestamp,
        [User name column] AS UserName,
        [Employee department column] AS EmployeeDepartment,
        [Report total amount column] AS ReportTotalAmount,
        [Expense category column] AS ExpenseCategory,
        [Policy violation flag column] AS PolicyViolationFlag,
        CAST(NULL AS VARCHAR(4000)) AS RevisionReason
    FROM [Your expense submission table]
    WHERE [Submission timestamp column] >= :start_date
      AND [Submission timestamp column] < :end_date
      AND ([Company column] = :company_filter OR :company_filter = '[Company filter]')
),
policy_check_performed AS (
    SELECT
        [Expense report ID column] AS ExpenseReportId,
        'Policy Check Performed' AS ActivityName,
        [Policy check timestamp column] AS EventTimestamp,
        [User name column] AS UserName,
        [Employee department column] AS EmployeeDepartment,
        [Report total amount column] AS ReportTotalAmount,
        [Expense category column] AS ExpenseCategory,
        [Policy violation flag column] AS PolicyViolationFlag,
        CAST(NULL AS VARCHAR(4000)) AS RevisionReason
    FROM [Your policy check table]
    WHERE [Policy check timestamp column] >= :start_date
      AND [Policy check timestamp column] < :end_date
      AND ([Company column] = :company_filter OR :company_filter = '[Company filter]')
),
manager_review_pending AS (
    SELECT
        [Expense report ID column] AS ExpenseReportId,
        'Manager Review Pending' AS ActivityName,
        [Manager pending timestamp column] AS EventTimestamp,
        [User name column] AS UserName,
        [Employee department column] AS EmployeeDepartment,
        [Report total amount column] AS ReportTotalAmount,
        [Expense category column] AS ExpenseCategory,
        [Policy violation flag column] AS PolicyViolationFlag,
        CAST(NULL AS VARCHAR(4000)) AS RevisionReason
    FROM [Your expense status history table]
    WHERE [Status column] = 'Pending Manager Approval'
      AND [Manager pending timestamp column] >= :start_date
      AND [Manager pending timestamp column] < :end_date
      AND ([Company column] = :company_filter OR :company_filter = '[Company filter]')
),
manager_approved AS (
    SELECT
        [Expense report ID column] AS ExpenseReportId,
        'Manager Approved' AS ActivityName,
        [Manager approval timestamp column] AS EventTimestamp,
        [User name column] AS UserName,
        [Employee department column] AS EmployeeDepartment,
        [Report total amount column] AS ReportTotalAmount,
        [Expense category column] AS ExpenseCategory,
        [Policy violation flag column] AS PolicyViolationFlag,
        CAST(NULL AS VARCHAR(4000)) AS RevisionReason
    FROM [Your manager approval table]
    WHERE [Manager approval timestamp column] >= :start_date
      AND [Manager approval timestamp column] < :end_date
      AND ([Company column] = :company_filter OR :company_filter = '[Company filter]')
),
finance_review_pending AS (
    SELECT
        [Expense report ID column] AS ExpenseReportId,
        'Finance Review Pending' AS ActivityName,
        [Finance pending timestamp column] AS EventTimestamp,
        [User name column] AS UserName,
        [Employee department column] AS EmployeeDepartment,
        [Report total amount column] AS ReportTotalAmount,
        [Expense category column] AS ExpenseCategory,
        [Policy violation flag column] AS PolicyViolationFlag,
        CAST(NULL AS VARCHAR(4000)) AS RevisionReason
    FROM [Your expense status history table]
    WHERE [Status column] = 'Pending Finance Approval'
      AND [Finance pending timestamp column] >= :start_date
      AND [Finance pending timestamp column] < :end_date
      AND ([Company column] = :company_filter OR :company_filter = '[Company filter]')
),
finance_approved AS (
    SELECT
        [Expense report ID column] AS ExpenseReportId,
        'Finance Approved' AS ActivityName,
        [Finance approval timestamp column] AS EventTimestamp,
        [User name column] AS UserName,
        [Employee department column] AS EmployeeDepartment,
        [Report total amount column] AS ReportTotalAmount,
        [Expense category column] AS ExpenseCategory,
        [Policy violation flag column] AS PolicyViolationFlag,
        CAST(NULL AS VARCHAR(4000)) AS RevisionReason
    FROM [Your finance approval table]
    WHERE [Finance approval timestamp column] >= :start_date
      AND [Finance approval timestamp column] < :end_date
      AND ([Company column] = :company_filter OR :company_filter = '[Company filter]')
),
expense_sent_back AS (
    SELECT
        [Expense report ID column] AS ExpenseReportId,
        'Expense Sent Back for Revision' AS ActivityName,
        [Revision timestamp column] AS EventTimestamp,
        [User name column] AS UserName,
        [Employee department column] AS EmployeeDepartment,
        [Report total amount column] AS ReportTotalAmount,
        [Expense category column] AS ExpenseCategory,
        [Policy violation flag column] AS PolicyViolationFlag,
        [Revision reason column] AS RevisionReason
    FROM [Your expense status history table]
    WHERE [Status column] IN ('Needs Revision', 'Rejected')
      AND [Revision timestamp column] >= :start_date
      AND [Revision timestamp column] < :end_date
      AND ([Company column] = :company_filter OR :company_filter = '[Company filter]')
),
reimbursement_scheduled AS (
    SELECT
        [Expense report ID column] AS ExpenseReportId,
        'Reimbursement Scheduled' AS ActivityName,
        [Reimbursement scheduled timestamp column] AS EventTimestamp,
        [User name column] AS UserName,
        [Employee department column] AS EmployeeDepartment,
        [Report total amount column] AS ReportTotalAmount,
        [Expense category column] AS ExpenseCategory,
        [Policy violation flag column] AS PolicyViolationFlag,
        CAST(NULL AS VARCHAR(4000)) AS RevisionReason
    FROM [Your reimbursement schedule table]
    WHERE [Reimbursement scheduled timestamp column] >= :start_date
      AND [Reimbursement scheduled timestamp column] < :end_date
      AND ([Company column] = :company_filter OR :company_filter = '[Company filter]')
),
reimbursement_executed AS (
    SELECT
        [Expense report ID column] AS ExpenseReportId,
        'Reimbursement Executed' AS ActivityName,
        [Reimbursement executed timestamp column] AS EventTimestamp,
        [User name column] AS UserName,
        [Employee department column] AS EmployeeDepartment,
        [Report total amount column] AS ReportTotalAmount,
        [Expense category column] AS ExpenseCategory,
        [Policy violation flag column] AS PolicyViolationFlag,
        CAST(NULL AS VARCHAR(4000)) AS RevisionReason
    FROM [Your reimbursement payment table]
    WHERE [Reimbursement executed timestamp column] >= :start_date
      AND [Reimbursement executed timestamp column] < :end_date
      AND ([Company column] = :company_filter OR :company_filter = '[Company filter]')
),
synced_with_accounting AS (
    SELECT
        [Expense report ID column] AS ExpenseReportId,
        'Synced with Accounting System' AS ActivityName,
        [Accounting sync timestamp column] AS EventTimestamp,
        [User name column] AS UserName,
        [Employee department column] AS EmployeeDepartment,
        [Report total amount column] AS ReportTotalAmount,
        [Expense category column] AS ExpenseCategory,
        [Policy violation flag column] AS PolicyViolationFlag,
        CAST(NULL AS VARCHAR(4000)) AS RevisionReason
    FROM [Your accounting synchronization table]
    WHERE [Accounting sync timestamp column] >= :start_date
      AND [Accounting sync timestamp column] < :end_date
      AND ([Company column] = :company_filter OR :company_filter = '[Company filter]')
)
SELECT ExpenseReportId, ActivityName, EventTimestamp, UserName, EmployeeDepartment, ReportTotalAmount, ExpenseCategory, PolicyViolationFlag, RevisionReason FROM expense_incurred
UNION ALL
SELECT ExpenseReportId, ActivityName, EventTimestamp, UserName, EmployeeDepartment, ReportTotalAmount, ExpenseCategory, PolicyViolationFlag, RevisionReason FROM receipt_attached
UNION ALL
SELECT ExpenseReportId, ActivityName, EventTimestamp, UserName, EmployeeDepartment, ReportTotalAmount, ExpenseCategory, PolicyViolationFlag, RevisionReason FROM expense_submitted
UNION ALL
SELECT ExpenseReportId, ActivityName, EventTimestamp, UserName, EmployeeDepartment, ReportTotalAmount, ExpenseCategory, PolicyViolationFlag, RevisionReason FROM policy_check_performed
UNION ALL
SELECT ExpenseReportId, ActivityName, EventTimestamp, UserName, EmployeeDepartment, ReportTotalAmount, ExpenseCategory, PolicyViolationFlag, RevisionReason FROM manager_review_pending
UNION ALL
SELECT ExpenseReportId, ActivityName, EventTimestamp, UserName, EmployeeDepartment, ReportTotalAmount, ExpenseCategory, PolicyViolationFlag, RevisionReason FROM manager_approved
UNION ALL
SELECT ExpenseReportId, ActivityName, EventTimestamp, UserName, EmployeeDepartment, ReportTotalAmount, ExpenseCategory, PolicyViolationFlag, RevisionReason FROM finance_review_pending
UNION ALL
SELECT ExpenseReportId, ActivityName, EventTimestamp, UserName, EmployeeDepartment, ReportTotalAmount, ExpenseCategory, PolicyViolationFlag, RevisionReason FROM finance_approved
UNION ALL
SELECT ExpenseReportId, ActivityName, EventTimestamp, UserName, EmployeeDepartment, ReportTotalAmount, ExpenseCategory, PolicyViolationFlag, RevisionReason FROM expense_sent_back
UNION ALL
SELECT ExpenseReportId, ActivityName, EventTimestamp, UserName, EmployeeDepartment, ReportTotalAmount, ExpenseCategory, PolicyViolationFlag, RevisionReason FROM reimbursement_scheduled
UNION ALL
SELECT ExpenseReportId, ActivityName, EventTimestamp, UserName, EmployeeDepartment, ReportTotalAmount, ExpenseCategory, PolicyViolationFlag, RevisionReason FROM reimbursement_executed
UNION ALL
SELECT ExpenseReportId, ActivityName, EventTimestamp, UserName, EmployeeDepartment, ReportTotalAmount, ExpenseCategory, PolicyViolationFlag, RevisionReason FROM synced_with_accounting
ORDER BY ExpenseReportId, EventTimestamp, ActivityName
""")

with engine.connect() as connection:
    event_log = pd.read_sql(
        sql,
        connection,
        params={
            class="hl-string">"start_date": START_DATE,
            class="hl-string">"end_date": END_DATE,
            class="hl-string">"company_filter": COMPANY_FILTER
        }
    )

event_log[class="hl-string">"EventTimestamp"] = pd.to_datetime(event_log[class="hl-string">"EventTimestamp"], errors=class="hl-string">"coerce", utc=True)
event_log = event_log[event_log[class="hl-string">"ExpenseReportId"].notna() & event_log[class="hl-string">"EventTimestamp"].notna()].copy()
event_log.to_csv(class="hl-string">"ramp_expense_event_log.csv", index=False)
print(event_log.head())
print(event_log.groupby(class="hl-string">"ActivityName", dropna=False).size())

准备好开始了吗？

利用此模板改进费用管理流程、提升效率并加快审批。立即开始优化运营。

返回改进指南下载模板

释放效率：立即优化Ramp费用管理

准确定位低效环节，简化审批流程，将周期时间缩短30%。

开始免费试用

无需信用卡，几分钟即可完成设置。

您的费用管理数据模板

您的费用管理数据模板

费用管理属性

费用管理活动

提取指南

Ramp内置CSV交易和报销导出

步骤

配置

a 示例查询 python

通过Ramp Developer API使用Python脚本提取数据

步骤

配置

a 示例查询 python

Ramp数据仓库集成同步（Snowflake/BigQuery）

步骤

配置

a 示例查询 python

准备好开始了吗？

释放效率：立即优化Ramp费用管理

您的费用管理数据模板

您的费用管理数据模板

费用管理属性

费用管理活动

提取指南

Ramp内置CSV交易和报销导出

步骤

配置

a 示例查询 python

通过Ramp Developer API使用Python脚本提取数据

步骤

配置

a 示例查询 python

Ramp数据仓库集成同步（Snowflake/BigQuery）

步骤

配置

a 示例查询 python

准备好开始了吗？

释放效率：立即优化Ramp费用管理

我们重视您的隐私