> 持续流程改进 > 费用管理 > 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	费用管理流程中某个时间点发生的具体事件或任务的名称。
说明此属性描述费用报告生命周期中的一个步骤，例如“已提交费用”“经理已批准”或“已完成报销付款”。这些活动构成流程图中的节点，用于呈现和分析流程。分析活动有助于找出最常发生的步骤、瓶颈所在，以及不同工作流之间的差异。这是了解操作顺序和衡量各阶段绩效的核心要素。重要性此属性定义流程图中的步骤，便于呈现和分析从开始到结束的流程。获取途径此属性通常根据Ramp中与每份费用报告相关的事件日志或状态变更记录生成。示例已提交费用经理已批准待财务审核已完成报销付款
	费用报告ID ExpenseReportId	每份费用报告的唯一标识符，也是该流程的主要案例标识符。
说明费用报告ID将单次费用提交相关的所有事件和活动归为一组。借助此ID，您可以按时间顺序完整跟踪费用申请，从最初录入到最终付款。在流程挖掘中，此属性对于还原每份费用报告的端到端过程至关重要。将其用作案例ID后，分析便能准确计算周期时间、发现瓶颈，并呈现报告在审批流程中的不同路径。重要性此属性是将所有相关活动关联到同一流程实例的基础，使端到端分析成为可能。获取途径此标识符应位于Ramp的主要费用报告表或交易表中。示例 ER-2023-08-1123ER-2023-09-4591ER-2023-10-0024
	修改原因 RevisionReason	费用报告退回员工修改时填写的原因。
说明审批人拒绝或退回费用报告时，通常会填写原因。此属性记录具体原因，例如“缺少收据”“类别错误”或“违反政策”。此信息对于“费用修改率与原因”仪表板非常有价值。分析最常见的返工原因，有助于组织发现提交流程中的系统性问题，并开展有针对性的培训或改进系统，以减少错误。重要性此属性直接揭示返工的根本原因，便于采取有针对性的措施，提高首次提交质量。获取途径在Ramp中发生“费用已退回修改”活动时，系统会在备注或拒绝详情中记录此数据。示例缺少分项收据费用超出政策限额选择了错误的费用类别重复交易
	员工部门 EmployeeDepartment	提交费用报告的员工所属部门。
说明此属性表示提交费用的员工所属业务单元或部门，例如“销售”“工程”或“市场营销”。这是重要的分析维度，可用于筛选和比较组织不同部门的流程绩效。它能揭示各部门在审批时间、退回修改率和政策合规方面的差异，为有针对性的改进提供依据。重要性此属性便于比较不同业务单元的流程指标，揭示效率、合规和支出方面的差异。获取途径此信息可能保存在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
	审批步骤数 ApprovalStepCount	费用报告经过的正式审批步骤总数。
说明此计算指标统计单份费用报告中发生的不同审批活动数量，例如“经理已批准”和“财务已批准”，用于量化审批工作流的复杂程度。此属性用于“每份报告的平均审批步骤数”KPI和“简化费用审批路径”仪表板。结合报告金额或类别分析审批步骤数，组织可以发现简单、低金额的费用是否经历了过于复杂的审批流程。重要性此属性用于量化工作流的复杂程度，帮助您发现简化流程的机会，尤其是针对低风险报告。获取途径此指标通过统计事件日志中每个案例的特定审批活动次数得出。示例 123
	审批经理 ApprovingManager	执行审批步骤的经理姓名。
说明此属性用于标识负责审核并批准员工费用报告的经理，与提交报告的用户不同。跟踪审批经理对于“经理审批周期时间”仪表板至关重要。您可以据此分析审批工作量和绩效，找出审批较快的经理以及造成瓶颈的经理，从而平衡工作量或提供额外支持。重要性此属性便于分析各审批人的绩效，帮助发现并解决审批工作流中的瓶颈。获取途径这项信息属于Ramp中的审批工作流数据，记录经理对报销报告采取操作时的情况。示例 Jane DoeJohn MillerSusan Chen
	审计结果 AuditOutcome	对费用报告进行内部或外部审计后得出的结果。
说明对于经过正式审计的费用报告，此属性记录最终结果，例如“已批准”“部分拒绝”或“需要补充信息”。这些数据是“费用审计绩效”仪表板的核心，可用于评估审计流程的有效性、跟踪各类结果的发生频率，并了解审计发现带来的财务影响。重要性此属性用于衡量审计流程的成效和结果，帮助您了解合规情况及控制环节的薄弱之处。获取途径如果使用Ramp的审计模块或集成系统进行详细的费用审计，审计结果会记录在相应系统中。示例通过通过，但有备注已拒绝已升级处理
	报销方式 ReimbursementMethod	用于执行报销付款的方式，例如ACH转账或电汇。
说明此属性用于说明员工通过哪种付款渠道获得报销。不同方式的处理时间和相关费用可能有所不同。按方式分析报销绩效，有助于找出效率最高的付款渠道。“报销方式绩效”仪表板利用这些数据比较周期时间和可靠性，为优化付款策略提供依据。重要性通过比较不同付款渠道的表现，您可以优化付款速度和可靠性。获取途径您可以在Ramp的付款或报销记录中找到这项信息。示例 ACH转账公司卡信用额度直接存款
	提交方式 SubmissionMethod	提交费用报告所用的渠道，例如移动应用或网页门户。
说明此属性用于说明员工通过何种方式提交费用报告。常见方式包括使用移动应用、桌面浏览器或转发电子邮件。分析提交方式有助于了解用户行为和技术使用情况。例如，通过某个渠道提交的报告修改率较高，可能表明该渠道的界面存在易用性问题。重要性此属性有助于了解用户行为，并判断某些提交渠道是否更容易出现错误或延误。获取途径这项信息可能记录在Ramp系统日志中提交事件的元数据里。示例移动应用网页门户电子邮件
	数据最近更新时间 LastDataUpdate	表示数据最近一次从源系统刷新时间的时间戳。
说明此属性记录最近一次数据提取的日期和时间，说明所分析数据的新旧程度。解读分析结果时，了解数据的更新时间至关重要。此属性有助于您判断当前查看的信息是否为最新数据。重要性此属性说明数据的时效性，确保分析基于当前且相关的信息。获取途径此时间戳在数据提取过程中生成并添加。示例 2023-11-01T06:00:00Z
	是否返工 IsRework	计算得出的标记，用于表示费用报告是否至少一次被退回修改。
说明此布尔属性通过检查某个案例是否发生过“费用报告退回修改”活动得出。凡是至少经历过一轮修改的报告，该属性均设为true。此标记简化了“费用报告修改率”KPI的计算，也便于筛选和比较首次提交即获批准的报告与需要返工的报告。分析这两类报告，可以了解修改对处理时间和成本的影响。重要性此属性便于按返工情况划分流程，帮助您量化报告被退回修改的频率及其影响。获取途径这是一个计算字段，在数据转换时通过检查案例中是否存在修改活动得出。示例 truefalse
	源系统 SourceSystem	标识提取数据的源应用程序。
说明此属性用于标明流程数据所属的记录系统，本例中为Ramp。在数据可能来自多个系统的环境中，它有助于明确数据来源。分析时，您可以按特定来源筛选数据，也可以将其用于数据验证和治理。重要性此属性说明数据的来源，这对于数据治理和整合多个系统的数据至关重要。获取途径此属性通常是在数据提取和转换过程中添加的固定值（“Ramp”）。示例 Ramp
	财务审批人 FinanceApprover	财务团队中负责批准费用报告的用户。
说明对于包含财务审核环节的工作流，此属性用于识别财务部门中执行最终审批的具体人员或团队。借助此属性，“财务审核效率”仪表板可以按个人或团队分析绩效，帮助您发现瓶颈、评估工作量分配，并衡量财务审核环节的成效。重要性此属性支持对财务审核环节进行详细的绩效分析，帮助您优化流程中的关键控制点。获取途径此信息会在Ramp费用报告的审批历史中记录，具体发生在财务审核环节。示例财务团队ADavid Lee财务自动化机器人

必填建议填写选填

费用管理活动

要准确发现流程并衡量绩效，请在事件日志中记录这些关键步骤和里程碑。

5 建议填写 7 选填

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

建议填写选填

数据提取指南

如何从Ramp获取数据

步骤

使用管理员或簿记员权限登录Ramp账户。
在左侧菜单中进入交易。此页面包含所有公司卡支出。
设置所需的数据筛选条件。至少要设置要分析的日期范围。如有需要，还可以按状态、类别或部门等其他属性筛选，缩小范围。
点击交易列表右上角的导出按钮。
在导出窗口中，将导出类型设为交易，格式设为CSV。下载文件并命名为transactions_export.csv。
接着，在左侧菜单中进入报销。此页面包含员工提交的所有自付费用。
使用与导出交易数据时相同的日期范围筛选条件，确保数据一致。
点击导出按钮，将报销数据下载为CSV文件，并命名为reimbursements_export.csv。
将导出的两个CSV文件（transactions_export.csv和reimbursements_export.csv）放在提供的Python脚本所在目录中。
运行Python脚本。脚本会读取两个文件，将每个业务事件转换为单独一行，并合并为一份结构化事件日志。
脚本会生成名为ramp_event_log.csv的文件，其格式符合ProcessMind的要求。
将ramp_event_log.csv上传到ProcessMind进行分析。

配置

权限：您必须在Ramp中拥有“管理员”或“簿记员”角色，才能访问并导出完整的交易和报销数据集。
日期范围：首次分析建议选择至少6至12个月的日期范围，以涵盖足够多的流程变化。持续监控时，可以选择较短且较新的时间范围。
数据一致性：导出“交易”和“报销”文件时，请使用完全相同的日期范围和筛选条件，以确保数据集一致。
导出格式：两个文件均应选择CSV格式，因为数据转换脚本专门用于处理逗号分隔文件。
文件命名：请使用transactions_export.csv和reimbursements_export.csv作为文件名，或相应修改数据转换脚本中的文件路径变量。

查询示例 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}'.")

步骤

准备工作和环境设置：确保已安装Python 3.8或更高版本。运行pip install requests pandas安装所需的Python库。您还需要Ramp实例的管理员权限，以生成API凭据。
生成Ramp API凭据：登录Ramp账户，进入“设置”，然后从侧边菜单中选择Developer API。点击“创建应用”，为应用命名并配置所需权限范围。您需要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进行身份验证，从用户、交易和报销端点获取数据，并自动处理分页。
了解API调用过程：脚本首先获取所有用户数据，建立用户ID与姓名、部门之间的映射。随后提取指定日期范围内的所有报销记录，并获取每笔报销的审批和付款历史。
将数据转换为事件日志：脚本的核心逻辑会遍历每笔报销及其关联的状态历史，并将“MANAGER_APPROVAL”或“FINISHED”等状态变更转换为“经理审核待处理”或“已执行报销”等独立活动。脚本还会获取相关交易数据，生成“发生费用”和“已附上收据”事件。
补充数据：脚本会为事件日志中的每项活动补充用户详情、部门信息和费用类别，生成适用于流程分析的完整数据集。
检查输出文件：脚本运行完成后，会在同一目录中生成ramp_event_log.csv。请使用电子表格程序打开文件，检查数据是否按预期提取。
准备上传：生成的CSV文件包含所需列ExpenseReportId、ActivityName、EventTimestamp及其他建议属性，可直接上传到ProcessMind平台。

配置

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

查询示例 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配置，请根据[您的系统配置]进行设置。
配置数据仓库目标位置、数据库或项目、架构或数据集、区域、凭据和同步范围。记录Ramp显示的复制表名和列名。请勿假设表名固定不变，因为它们会因集成方式和账户配置而异。
找出已复制的费用报告、费用交易、收据、审批、政策、报销和会计同步数据。将每个数据源映射到事件日志所需字段。如果某个必要的生命周期事件没有明确记录，请找出记录该事件的源状态历史或审计数据。
使用[开始日期]和[结束日期]定义提取时间范围，建议选择三至六个月。只有在存在相应源字段时，才应用公司或组织筛选条件。保留源时区，或将所有时间戳转换为已记录的统一时区。
创建数据仓库视图或查询，将源字段规范化为ExpenseReportId、ActivityName、EventTimestamp、UserName、EmployeeDepartment、ReportTotalAmount、ExpenseCategory、PolicyViolationFlag和RevisionReason。请将所提供查询中的所有方括号表名和列名占位符替换为复制后架构中的实际名称。
将每项生命周期活动提取为明确的事件行。下方查询针对12项必需活动分别包含一个完整的SELECT代码块，不会根据无关记录推断事件。对于基于状态的活动，请使用源状态历史或事件表，而不只是当前报告状态，以保留历史状态变更。
验证标识符、时间戳、活动标签、重复记录处理方式和空值处理方式。确认每个事件都属于正确的ExpenseReportId，并在源数据记录了多个有效事件时保留重复活动。
将结果导出为CSV或ProcessMind支持的其他表格格式。请使用准确的必需列名ExpenseReportId、ActivityName和EventTimestamp，并在可用时加入建议属性。上传文件时，将“费用报告ID”设为案例标识符。ProcessMind会按原样读取事件日志，不会自动补全缺失的活动。

配置

集成目标位置：为Snowflake或BigQuery配置Ramp原生数据仓库集成，并根据[您的系统配置]完成账户专属设置。
源对象：使用数据仓库中提供的已复制费用报告、交易、收据、审批、政策、报销和会计同步对象。请将[您的表名]和[您的列名]占位符替换为已核实的集成对象名称。
日期范围：先提取三至六个月的数据。验证数据仓库成本、查询时长、事件量和ProcessMind上传限制后，再扩大时间范围。
日期筛选：如果目标是捕获该时间段内发生的所有活动，请按事件时间戳筛选，而不只是按报告创建时间筛选。
组织筛选：仅当复制数据中包含公司代码、实体、部门、卡计划或其他相关字段时，才应用相应筛选条件。如需限制到特定公司，请使用[公司筛选条件]。
活动历史：优先使用不可变的审计记录或状态历史记录。仅使用当前状态表可能无法保留收据附件、提交、审批、修改、报销或会计同步的历史记录。
时间戳处理：将时间戳统一为同一时区，并记录源数据使用UTC还是本地时间。尽可能保留源数据的时间精度。
所需访问权限：管理员可能需要配置集成、创建或查询数据仓库对象，以及访问费用、审批、报销和会计数据的权限。
许可与可用性：数据仓库复制、历史审计数据、报销数据和会计同步数据是否可用，取决于Ramp方案、已启用模块、集成配置和数据保留政策。
性能：尽早缩小日期范围，只选择必需列，并在适当时按组织筛选。对于大型数据集，可将中间的规范化视图或表实体化。根据目标数据仓库支持的功能进行分区或聚簇。
增量运行：定期加载时，以已处理的最大EventTimestamp加一段较短的重叠时间作为水位标记。只有在确认源事件标识符可靠后才进行去重，因为删除重复事件可能改变流程表现。

查询示例 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文件

步骤

配置

查询示例 python

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

步骤

配置

查询示例 python

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

步骤

配置

查询示例 python

准备好开始了吗？

提升效率：立即优化Ramp费用管理