您的患者旅程数据模板

Athenahealth

← 返回改进指南

患者旅程属性 (20) 患者旅程活动 (15) 提取指南 (2)

您的患者旅程数据模板

此模板为收集分析患者旅程所需的必要数据提供了全面指南。它概述了您需要追踪的关键属性和活动，并就如何从 athenahealth 提取这些信息提供了明确指导。利用此资源确保您的数据已准备好进行有效的流程挖掘。

建议收集的属性
需要追踪的关键活动
athenahealth 提取指南

不熟悉事件日志？了解如何创建流程挖掘事件日志.

下载模板

患者旅程属性

这些是推荐包含在事件日志中的数据字段，对于全面分析和理解患者旅程至关重要。

5 必填 10 推荐 5 可选

	名称	描述
	事件timestamp EventTimestamp	活动发生的具体日期和时间。
描述记录活动发生的准确时刻。用于按时间顺序排列事件并计算步骤之间的时长。对于基于时间的分析（包括周期时间、等待时间和吞吐量分析）至关重要。建议使用高精度时间戳，以区分同一天内发生的事件顺序。为何重要用于事件排序和计算所有基于时间的 KPI。获取方式与 athenahealth 表中状态更改或创建日期相关的时间戳字段。示例 2023-10-12T08:30:00Z2023-10-12T09:15:22Z2023-10-15T14:20:00Z
	最后数据更新 LastDataUpdate	数据上次提取或刷新的时间戳。
描述指示分析中所用数据的时效性。这有助于用户了解他们查看的是实时数据还是历史快照。用于管理数据流水线，并确保仪表板反映流程的最新状态。为何重要对于数据治理以及用户对仪表板实时性的信任至关重要。获取方式 ETL 执行时刻的系统时间。示例 2023-11-01T12:00:00Z2023-11-02T06:00:00Z
	患者就诊阶段 PatientEpisodeId	特定患者接诊阶段或旅程的唯一标识符。
描述此属性用作核心案例标识符 (Case Identifier)，将与患者特定诊疗阶段或病情相关的所有活动分组。它将预约、诊断医嘱和出院程序等分散事件连接成一个连贯的旅程。在分析中，此 ID 是流程挖掘的主键，用于重构端到端流程。它确保同一位患者因不同病情而进行的多次就诊被视为不同的流程实例。为何重要对于定义分析中单个流程实例的范围至关重要。获取方式源自 athenahealth 中的就诊 ID 分组或关联到特定的“护理病程 (Episode of Care) ID”。示例 EP-2023-88491EP-2023-99102ENC-55412-GRP
	活动名称 ActivityName	在患者旅程中执行的事件或任务的名称。
描述指示流程中发生的具体步骤，如“患者已签到”、“已下达诊断测试医嘱”或“已给药”。此字符串定义了流程图中的节点。用于将流程可视化并识别操作顺序。这些名称的标准化对于生成清晰、可读的流程图至关重要。为何重要定义流程图的步骤，是任何流程挖掘的必填项。获取方式源自审计日志、预约状态变更或报销单行项目描述。示例已预约会议患者已签到诊断测试已下达医嘱患者已出院
	源系统 SourceSystem	数据来源系统的名称。
描述识别负责生成记录的 IT 系统（在此案例中为 'athenahealth'）。这在数据可能混合的多系统环境中特别有用。允许分析师按数据源过滤视图，并针对特定系统排查数据质量问题。为何重要在多系统流程挖掘配置中提供数据血缘和上下文。获取方式硬编码的字面量或系统配置 ID。示例 AthenahealthAthenaOneAthenaPractice
	主诊断代码 PrimaryDiagnosisCode	与本次接诊阶段相关的主要 ICD-10 代码。
描述对患者旅程的临床原因进行分类。这为“年龄段与诊断旅程对比”提供了必要的背景信息。分析师以此按病情（如肺炎 vs. 骨折）对旅程进行细分，因为不同病情的预期路径和周期时间差异巨大。为何重要支持“同类”案例对比；不同诊断的周期时间差异巨大。获取方式就诊或报销诊断字段 (ICD-10)。示例 J18.9I10E11.9
	事件结束时间 EventEndTime	特定活动完成的时间。
描述捕获活动的完成时间，从而能够计算该步骤本身的活跃持续时间（处理时间），这与此前的等待时间不同。用于分析资源效率，并识别手动执行时间超过预期的任务。为何重要支持计算主动处理时间与被动等待时间的对比。获取方式 athenahealth 中的结账时间、结果验证时间或特定完成时间戳。示例 2023-10-12T09:00:00Z2023-10-12T10:45:00Z
	出院去向 DischargeDisposition	患者出院时的去向或状态。
描述指示患者在本次就诊后的去向，如“回家”、“专业护理机构”或“临终关怀”。这对于“出院计划与再入院趋势”至关重要。它提供了所需出院计划复杂性的背景信息，并有助于评估护理转接的有效性。为何重要出院计划和再入院风险的关键背景信息。获取方式 athenahealth 就诊字段或医院出院记录。示例主页专业护理机构转院至短期医院已到期
	医疗提供者姓名 ProviderName	执行该活动的医疗专业人员姓名。
描述识别负责该事件的特定医生、护士或技术人员。此属性是资源利用分析的核心。它允许比较不同工作人员的性能指标（如吞吐量和周期时间），以识别培训需求或工作量不平衡问题。为何重要 “资源利用率”和交接分析的关键。获取方式解析为服务提供者目录名称的 athenahealth 'providerid'。示例史密斯医生护士琼斯技术员亚当斯
	处理时间 ProcessingTime	积极处理该活动所花费的时长。
描述代表特定任务从开始到完成的时间差（例如，执行扫描所需的时间）。这对应于通用模型中的“ProcessingTime”。用于计算资源利用率，并识别手动任务与自动化任务之间的效率差距。为何重要在总周期中区分主动工作时间与等待时间。获取方式计算：EventEndTime - EventTimestamp。示例 15分钟1小时30分钟45秒
	就诊类型 EncounterType	就诊类别（例如诊室就诊、远程医疗、急诊）。
描述定义所提供护理的形式或环境。这在通用数据模型中充当“案例类型 (Case Type)”。不同的就诊类型有不同的预期流程和计费要求。按此属性进行过滤对于避免周期时间分析中的“苹果比橘子”（不具可比性的对比）至关重要。为何重要区分不同的流程变体，如远程医疗 vs. 线下就诊。获取方式 athenahealth 'encountertype' 字段。示例诊室就诊远程医疗紧急手术
	患者 ID PatientId	患者的唯一标识符（已匿名化/哈希处理）。
描述唯一标识正在进行旅程的客户（患者）。虽然与案例标识符相似，但一名患者可能会随着时间的推移有多次就诊。用于关联重复就诊并分析再入院率。这对于“计划外再入院百分比”KPI 至关重要。为何重要用于跨就诊阶段追踪再入院情况和患者病史。获取方式 athenahealth 'patientid' 字段。示例 PAT-100234PAT-559201PAT-992210
	患者年龄组 PatientAgeGroup	患者年龄的分组类别（如 18-25、65+）。
描述将患者划分为人口统计群体。这是“年龄组与诊断旅程比较”仪表板直接需要的。它有助于识别流程效率低下或成效是否不成比例地影响特定的年龄群体，如老年人或儿科患者。为何重要用于医疗流程分析的标准人口统计细分。获取方式根据患者出生日期和开始时间推导。示例 18-2930-4965岁以上
	是否再入院 IsReadmission	指示此病程是否代表非计划返回就诊的标记。
描述一个布尔指标，标识患者是否在之前出院后的设定时间段内（如 30 天）返回医院。这直接支持“非计划再入院百分比”KPI。通过过滤此属性，分析人员可以深入挖掘再入院的根本原因，并识别初始治疗路径中的模式。为何重要直接支持再入院率 KPI。获取方式在 ETL 过程中通过比较入院日期与之前的出院日期来计算。示例 truefalse
	部门名称 DepartmentName	发生该活动的医院或诊所部门。
描述按职能单元（如急诊、心脏科或放射科）对流程数据进行细分。这对于“部门吞吐量与交接”仪表板至关重要。使用此属性进行分析可以突显特定区域的瓶颈，并有助于优化跨部门的患者流。为何重要对于识别组织瓶颈和交接环节的低效至关重要。获取方式解析为科室名称的 athenahealth 'departmentid'。示例急诊科内科放射科
	总收费金额 TotalChargeAmount	为该活动或接诊阶段收取的货币金额。
描述与活动相关的财务价值或总理赔金额。这实现了基于成本的流程挖掘和财务影响分析。用于识别治疗路径中的高成本偏差，并将流程效率与财务成效相关联。为何重要为流程分析增加了财务维度。获取方式 athenahealth 报销/费用表中的 'amount' 或 'totalcharge'。示例 150.002500.5045.00
	是否返工 IsRework	指示此活动是否为重复的标记。
描述一个布尔标记，如果该活动在同一个案例中发生多次，则其值为 true。这为“活动频率与返工循环”仪表板提供了支持。它能够立即隔离包含返工的案例，从而支持“返工活动频率”这一 KPI 的计算。为何重要识别流程效率低下和冗余步骤。获取方式在 ETL 过程中根据每个 CaseId 的活动发生情况计算。示例 truefalse
	索赔状态 ClaimStatus	与诊疗相关的财务理赔状态。
描述指示账单理赔的状态，如“已提交”、“已拒绝”或“已支付”。这与“理赔已提交”活动和财务吞吐量相关。有助于识别临床记录问题是否导致了后端财务延迟。为何重要将临床效率与收入周期绩效联系起来。获取方式 athenahealth 'claimstatus' 字段。示例已开账单挂起作废
	订单类型 OrderType	医嘱类别（如化验、影像、处方）。
描述对病程中下达的临床医嘱进行分类。这对于“诊断测试提前期”仪表板至关重要。它允许分析师专门测量化验与影像的提前期，因为这两者通常具有不同的服务水平协议 (SLA) 和瓶颈。为何重要对诊断流程进行分段，以进行特定的提前期分析。获取方式医嘱 API 中的 athenahealth 'ordertype' 或 'class'。示例化验室影像处方处置/手术
	诊断结果状态 DiagnosticResultStatus	诊断医嘱的结果状态（例如阳性、正常）。
描述捕获测试的高级结果。这为“诊断测试提前期”和随后的治疗决策提供了背景。用于分析异常结果是否比正常结果更能促成更快的后续行动。为何重要将流程流向与临床结果关联起来。获取方式 athenahealth 实验室结果观察字段。示例普通异常严重

必填推荐可选

患者旅程活动

这些是事件日志中需要捕捉的关键流程步骤和里程碑，旨在准确发现并优化患者体验。

7 推荐 8 可选

	活动	描述
	出院医嘱已下达	医生或授权提供者下达正式医嘱，批准患者出院。这是在电子病历系统的 CPOE（医生医嘱录入）模块中创建的带有时间戳的明确事件。
为何重要此活动启动出院流程。该医嘱与实际出院之间的时间间隔是“出院计划提前期”的关键绩效指标。获取方式位于医嘱表中。该事件通过特定的“出院”医嘱类型及其创建时间戳来识别。捕获在医嘱表中创建了一条医嘱类型为“出院”的新记录。事件类型 explicit
	初始评估已完成	标志着首次临床评估（如分诊或护理评估）的完成，在此过程中记录了生命体征和主诉。此事件通常根据本次就诊的第一个签署的临床记录或完成的评估表的时间戳推断。
为何重要此里程碑标志着临床诊疗的开始。从签到该活动的持续时间是衡量患者初始等待时间和资源响应能力的关键指标。获取方式根据 athenaClinicals 中特定临床文档或流程图的创建或签署时间戳推断。需要识别分诊或摄入相关的文档类型。捕获识别本次接诊中临床记录、生命体征流程图或特定摄入表单上的第一个时间戳。事件类型 inferred
	已执行处置	对患者执行临床操作，如手术或专业治疗。这是临床文档中捕获的一个明确事件，通常在操作记录中记有具体的开始和结束时间。
为何重要医疗处置是患者治疗中的重要里程碑。分析处置前后的活动有助于优化术前和术后工作流。获取方式在 athenaClinicals 的医疗操作记录或特定的临床流程单中可以找到。事件时间戳源自记录的操作开始或结束时间。捕获创建了操作记录或日志，其中包含该操作发生的时间戳。事件类型 explicit
	患者已出院	患者已正式出院，其医疗机构内的旅程部分已结束。这是住院接诊的最后一个 ADT 事件，带有精确的时间戳。
为何重要此事件标志着患者主要旅程的结束。它是衡量整体“患者旅程周期时间”的终点，对于再入院分析至关重要。获取方式这是 ADT 系统或患者就诊表中的明确事件，标志着就诊的最终状态为“已出院”并带有时间戳。捕获患者就诊状态更新为“已出院”，并记录一条 ADT 事件。事件类型 explicit
	患者已签到	此活动标志着患者到达并为其预约的就诊正式签到。这通常捕捉为 athenaClinicals 或 athenaCommunicator 内预约记录上的明确状态更改。
为何重要这是患者现场旅程的明确起点。它是衡量等待时间和临床就诊整体周期时间的关键起点。获取方式记录为预约表或接诊表中的状态更新。查找“已签到”状态及其对应的时间戳。捕获预约或就诊对象的状态更改被记录并附带时间戳。事件类型 explicit
	测试结果已接收	诊断测试结果已定稿并录入患者病历。这通常发生在化验或影像系统将结果发回 athenahealth 并创建带时间戳的条目时。
为何重要结果的接收是后续临床决策（如诊断和治疗计划）的关键触发点。此事件是衡量诊断周转时间的终点。获取方式位于结果或诊断表中，链接到原始医嘱。该事件以结果归档或录入患者病历的时间戳为标志。捕获创建了带有时间戳的新结果记录，通常通过 LIS（实验室信息系统）或 RIS（放射信息系统）的接口生成。事件类型 explicit
	诊断已确认	临床医生正式为患者当前就诊的状况分配或确认诊断。这可以通过与患者就诊关联的主诊断代码（如 ICD-10）的创建或“最后更新”时间戳来推断。
为何重要这是一个决定后续治疗路径的关键里程碑。分析此点之后活动的偏差有助于理解和标准化护理协议。获取方式根据患者的问题列表或接诊的诊断数据推断。诊断代码的更改或定稿及其相关时间戳标志着该事件的发生。捕获检测就诊主诊断被添加或更新的时间戳。事件类型 inferred
	已预约会议	代表为患者预订预约。当用户在 athenahealth 的排程模块 athenaCommunicator 中创建并确认新的预约条目时，系统会明确捕捉到此事件。
为何重要此活动标志着许多患者旅程的初始接触点。分析从预约到签到的时间有助于了解患者就医的可及性和就诊前的效率。获取方式这是记录在预约或排程表中的明确事件。它通常与创建时间戳和患者 ID 相关联。捕获在排班模块中创建预约记录时记录该事件。事件类型 explicit
	患者已转院/转科	患者从一个护理单元或部门移至另一个，例如从急诊科移至住院病房。这通过电子健康记录 (EHR) 中的入院、出院、转院 (ADT) 事件明确捕捉。
为何重要此活动对于分析医疗机构内的部门交接和患者流至关重要。它有助于识别“患者交接时间”和资源配置中的瓶颈。获取方式记录在 ADT 事件日志或患者追踪表中。每个转运事件包含患者信息、起始/目标地点以及时间戳。捕获在患者转诊时生成带有时间戳的 ADT 消息或事件日志条目。事件类型 explicit
	标本已采集	代表从患者身上采集生物样本（如血液或尿液）进行化验的事件。这通常是记录在化验模块中的明确事件，或作为医嘱状态的更新。
为何重要这是诊断测试流程中的一个关键里程碑。医嘱、采集和结果之间的时间间隔可以揭示化验工作流中的重大瓶颈。获取方式此事件可以在化验医嘱的状态更改中找到，或者在与 athenahealth 对接的实验室信息系统中作为独立事件找到。查找“已采集”状态及其时间戳。捕获记录为化验医嘱状态的更新，或记录在专门的标本追踪模块中。事件类型 explicit
	治疗计划已制定	代表临床医生正式创建和记录患者的治疗计划。这可以捕捉为特定“护理计划”文档的创建或签署，或一系列相关的治疗医嘱。
为何重要此活动使预期的临床路径正式化。它是衡量标准协议合规性和分析诊疗偏差的关键点。获取方式通常存在于临床文档或医嘱表中。此事件对应于签署的护理计划记录或提交的一套协调治疗医嘱的时间戳。捕获事件是特定治疗计划文档或医嘱集的创建或定稿时间戳。事件类型 explicit
	索赔已提交	为患者就诊期间提供的服务生成报销单并提交给支付方。这是 athenaCollector 收入周期管理模块中的一个明确事件。
为何重要虽然这是一个行政步骤，但此活动对于分析与临床旅程并行的收入周期流程至关重要。它有助于识别临床诊疗与计费之间的延迟。获取方式可在报销或计费表中找到。该事件以报销单创建或提交时间戳为标记。捕获创建报销记录，并将其状态更新为“已提交”，同时附带时间戳。事件类型 explicit
	药物已施用	临床工作人员实际为患者配发药物。这在 athenaClinicals 的用药记录 (MAR) 模块中有明确记录，每个剂量都有精确的时间戳。
为何重要此活动代表直接的治疗干预。分析其时间点有助于衡量“首次治疗时间”并确保遵循给药时间表。获取方式可在 MAR 数据表中找到。每个配发事件都有患者 ID、药物 ID、剂量和配发时间戳。捕获每当记录药物已配发时，用药记录 (MAR) 中都会创建一个带有时间戳的记录。事件类型 explicit
	诊断测试已下达医嘱	医疗提供者为诊断测试（如化验、影像扫描或其他操作）下达医嘱。这是通过 athenaClinicals 中的 CPOE 功能创建的带有时间戳的明确事件。
为何重要这是一个触发诊断子流程的关键决策点。追踪此活动对于分析从医嘱到结果的“诊断测试提前期”KPI 至关重要。获取方式可在医嘱表中找到。每项医嘱都有患者标识符、医嘱名称、医嘱状态和创建时间戳。捕获在系统的医嘱表中创建了带有时间戳的新记录。事件类型 explicit
	随访预约已安排	在患者主要治疗或出院后为其安排随访预约。当在 athenaCommunicator 排班模块中创建新预约时，会明确捕获此事件。
为何重要此活动是理解出院后护理协调及其对再入院率等成效影响的关键。它展示了护理的连续性。获取方式记录在预约表中。该事件通过在出院日期之后发生的预约创建时间戳来识别。捕获在排班系统中创建了新的预约记录。事件类型 explicit

推荐可选

提取指南

如何从 athenahealth 获取数据

步骤

使用您的安全凭据登录 athenaNet 门户。
通过主菜单导航至“报表”模块，通常位于“Reports（报表）”>“Report Builder（报表生成器）”或“Custom Reports（自定义报表）”下。
点击“Create New Report（创建新报表）”，选择“Encounters（就诊）”或“Patient Visit（患者访问）”作为主要报表类别。这能确保粒度数据围绕特定患者病程展开。
在“Columns（列）”或“Fields（字段）”选择区域，搜索并添加“Patient ID（患者 ID）”和“Encounter ID（就诊 ID）”作为您的案例标识符 (Case ID)。
添加属性所需的统计学和上下文列，例如科室、提供者姓名、患者年龄和主诊断代码。
添加与活动相对应的特定 timestamp 列。您可能需要在不同的字段组中查找（例如：“Scheduling（排班）”用于预约已安排、“Check-in（签到）”用于签到时间、“Billing（计费）”用于报销单已提交）。确保选择“Date and Time（日期和时间）”格式，而不仅仅是日期。
对报表配置应用过滤器。将“Encounter Date（就诊日期）”范围设置为所需的分析窗口（例如：过去 6 个月），并过滤掉“已取消”或“未爽约”的预约，以关注已完成的旅程。
点击“Run Report（运行报表）”预览数据。验证各列包含数据且不全为空值。
选择“Export（导出）”选项，格式选择“CSV”（逗号分隔值）。不要使用 Excel/XLSX 的格式化功能，因为它们会增加处理难度。
将文件本地保存为 'athena_raw_export.csv'。
由于报表导出的格式是每次就诊一行且带有多个日期列（宽格式），您必须将其重构为事件日志（长格式）。使用“查询”部分提供的 Python 脚本执行此转换。
运行脚本生成最终的 'process_mining_event_log.csv'，并将此文件上传到 ProcessMind。

配置

报表类别：选择“就诊活动”或“临床信息”作为基础类别，以捕获大部分临床 workflow 步骤。
日期范围：将初始提取限制在 3-6 个月内，以避免报表超时或行数限制（标准视图通常上限约为 50,000 行）。
列选择：确保选择明确的 timestamp 字段。优先选择绝对的“签到日期/时间”，避免使用类似“签到后天数”的计算字段。
空值处理：将报表配置为针对缺失数据显显示为空白单元格，而非“N/A”等自定义文本，以便于脚本处理。
科室过滤：如果组织规模较大，请按特定的服务地点或科室（如“心脏科”、“骨科”）进行过滤，以保持数据集的针对性。
权限：用户需拥有 athenaNet 中的“报表创建者”或“临床报表”权限才能访问自定义报表生成器。

a 查询示例 config

import pandas as pd
import numpy as np
from datetime import datetime

# Configuration: Input and Output filenames
INPUT_FILE = 'athena_raw_export.csv'
OUTPUT_FILE = 'patient_journey_event_log.csv'

# Load the raw wide-format report from athenaNet
# Assumes standard CSV export. Adjust encoding if necessary (e.g., 'utf-8-sig' or 'latin1')
df = pd.read_csv(INPUT_FILE, dtype=str)

# Define the mapping between Activity Names and the columns in your athenaNet report
# You must ensure these column names match your Report Builder export headers exactly
activity_mapping = {
    'Appointment Scheduled': 'APPT_CREATED_TIMESTAMP',
    'Patient Checked In': 'CHECKIN_TIMESTAMP',
    'Initial Assessment Completed': 'INTAKE_COMPLETED_TIMESTAMP',
    'Diagnostic Test Ordered': 'ORDER_ENTRY_TIMESTAMP',
    'Specimen Collected': 'LAB_COLLECTION_TIMESTAMP',
    'Test Results Received': 'LAB_RESULT_TIMESTAMP',
    'Diagnosis Confirmed': 'DIAGNOSIS_ENTRY_TIMESTAMP',
    'Treatment Plan Developed': 'PLAN_NOTE_TIMESTAMP',
    'Medication Administered': 'MED_ADMIN_TIMESTAMP',
    'Procedure Performed': 'PROCEDURE_END_TIMESTAMP',
    'Patient Transferred': 'TRANSFER_TIMESTAMP',
    'Discharge Order Written': 'DISCHARGE_ORDER_TIMESTAMP',
    'Patient Discharged': 'DISCHARGE_TIMESTAMP',
    'Follow-Up Appointment Scheduled': 'NEXT_APPT_CREATED_TIMESTAMP',
    'Claim Submitted': 'CLAIM_SUBMISSION_TIMESTAMP'
}

# Define attribute columns to carry over to every event row
attribute_columns = {
    'ENCOUNTER_ID': 'PatientEpisodeId',
    'PATIENT_ID': 'PatientId',
    'PROVIDER_NAME': 'ProviderName',
    'DEPARTMENT_NAME': 'DepartmentName',
    'PRIMARY_DIAGNOSIS_CODE': 'PrimaryDiagnosisCode',
    'PATIENT_AGE': 'PatientAgeGroup',
    'ENCOUNTER_TYPE': 'EncounterType',
    'DISCHARGE_DISPOSITION': 'DischargeDisposition',
    'IS_READMISSION': 'IsReadmission'
}

# Initialize list to hold event rows
event_log_data = []

# Transformation Loop
for index, row in df.iterrows():
    episode_id = row.get('ENCOUNTER_ID')
    
    # Skip rows without a valid identifier
    if pd.isna(episode_id) or str(episode_id).strip() == '':
        continue

    # Iterate through each defined activity
    for activity_name, col_name in activity_mapping.items():
        timestamp_val = row.get(col_name)
        
        # Only create an event if a valid timestamp exists
        if pd.notna(timestamp_val) and str(timestamp_val).strip() != '':
            
            # Create base event object
            event = {
                'PatientEpisodeId': episode_id,
                'ActivityName': activity_name,
                'EventTimestamp': timestamp_val,
                'SourceSystem': 'athenahealth',
                'LastDataUpdate': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            }
            
            # Add attributes
            for csv_col, target_col in attribute_columns.items():
                event[target_col] = row.get(csv_col, None)
            
            event_log_data.append(event)

# Create DataFrame from the list of events
event_log_df = pd.DataFrame(event_log_data)

# Standardize Date Format (ISO 8601 Recommended)
# Adjust the format string matches the raw CSV format (e.g., '%m/%d/%Y %H:%M')
event_log_df['EventTimestamp'] = pd.to_datetime(event_log_df['EventTimestamp'], errors='coerce')

# Remove rows where timestamp parsing failed
event_log_df = event_log_df.dropna(subset=['EventTimestamp'])

# Sort by Episode and Time
event_log_df = event_log_df.sort_values(by=['PatientEpisodeId', 'EventTimestamp'])

# Output to CSV
event_log_df.to_csv(OUTPUT_FILE, index=False)

print(f"Transformation complete. Processed {len(df)} encounters into {len(event_log_df)} events.")
print(f"File saved to: {OUTPUT_FILE}")

步骤

注册 API 应用：登录 athenahealth 开发者门户并注册新应用，以获取 Client ID 和 Client Secret。确保应用拥有 Appointments（预约）、Clinicals (Charts)（临床/病历）、Orders（医嘱）和 Billing (Claims)（计费/报销）的权限范围。
配置 Python 环境：在执行环境中使用 pip install requests pandas 安装所需的 Python 库（requests, pandas）。
身份验证：实现 OAuth 2.0 客户端凭据授予流以检索访问令牌。该令牌必须包含在随后所有 API 调用的 Authorization 标头（Bearer 令牌）中。
定义范围：设置 Practice ID 和提取的日期范围。Athenahealth API 是特定于医疗机构的，因此如果您管理多个机构，必须遍历 Practice ID。
提取预约：查询 /appointments/booked 或 /appointments/changed 端点以检索基础就诊数据。将预约 ID 映射到 PatientEpisodeId。从创建日期中提取“预约已安排”，从签到时间中提取“患者已签到”。
遍历就诊以获取临床事件：针对每个预约 ID（就诊），查询临床子端点。调用 /chart/{encounterId}/vitals 获取“初始评估已完成”；调用 /chart/{encounterId}/orders 获取“诊断测试已下达”、“样本已采集”（基于状态）和“出院医嘱已下达”。
获取结果和诊断：查询 /chart/{encounterId}/analyteresults 或 /chart/{encounterId}/labresults 获取“测试结果已收到”；查询 /chart/{encounterId}/diagnoses 获取“诊断已确认”。
提取治疗和操作：查询 /chart/{encounterId}/plan 获取“治疗计划已制定”；查询 /chart/{encounterId}/procedures 获取“医疗操作已执行”；查询 /chart/{encounterId}/medications/admin 获取“药物已配发”。
识别 ADT 和报销：使用预约状态历史记录或特定医院端点（如果可用）来捕获“患者已转诊”和“患者已出院”。查询与预约相关的 /claims 以提取“报销单已提交”。
关联随访：通过在就诊时间范围内搜索为同一患者 ID 创建的后续预约，识别“随访预约已安排”。
转换与扁平化：将嵌套的 JSON 响应解析为扁平的事件字典列表。将所有时间戳转换为 ISO 8601 格式。将字段映射到 ActivityName、PatientEpisodeId 和 EventTimestamp。
导出数据：将字典列表加载到 pandas DataFrame 中，执行最终清洗（处理空值），并导出为可供 ProcessMind 接入的 CSV 文件。

配置

API 凭证：需要从 athenahealth 开发者门户获取具有生产访问权限的有效 Client ID 和 Secret。
Practice ID：医疗机构上下文的唯一数字标识符。如果是为医疗网络提取数据，必须按每个机构运行脚本。
日期范围：由于临床数据粒度较细、数据量大，建议提取窗口为 3 到 6 个月。
速率限制：athenaOne API 有严格的速率限制（例如每秒 5 次调用）。脚本包含休眠函数，但企业级提取应实施指数退避算法。
权限范围 (Scopes)：确保令牌具有 athena/service/Athenanet.MDP.* 的作用域，或针对预约、病历和财务数据的特定细粒度作用域。
分页：大多数端点返回的记录数量有限（默认 1000 条）。脚本必须处理 next 或 offset 参数。
科室筛选：如果分析特定流程（如急诊 vs. 门诊），请在预约获取阶段按科室 ID 进行过滤。

a 查询示例 python

import requests
import pandas as pd
import time
from datetime import datetime, timedelta
import base64

class=class="hl-string">"hl-comment"># Configuration
API_KEY = class="hl-string">'YOUR_CLIENT_ID'
API_SECRET = class="hl-string">'YOUR_CLIENT_SECRET'
PRACTICE_ID = class="hl-string">'195900'  class=class="hl-string">"hl-comment"># Replace with your Practice ID
BASE_URL = class="hl-string">'https://api.athenahealth.com/preview1'  class=class="hl-string">"hl-comment"># Use production URL for live data
START_DATE = class="hl-string">'01/01/2023'
END_DATE = class="hl-string">'01/31/2023'
OUTPUT_FILE = class="hl-string">'athena_patient_journey.csv'

def get_access_token():
    url = fclass="hl-string">'{BASE_URL}/oauth2/v1/token'
    auth_str = fclass="hl-string">'{API_KEY}:{API_SECRET}'
    b64_auth = base64.b64encode(auth_str.encode()).decode()
    headers = {
        class="hl-string">'Authorization': fclass="hl-string">'Basic {b64_auth}',
        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">'scope': class="hl-string">'athena/service/Athenanet.MDP.*'}
    response = requests.post(url, headers=headers, data=data)
    if response.status_code == 200:
        return response.json()[class="hl-string">'access_token']
    else:
        raise Exception(fclass="hl-string">'Auth failed: {response.text}')

def fetch_paginated(url, headers, params):
    results = []
    while True:
        response = requests.get(url, headers=headers, params=params)
        if response.status_code == 429:
            time.sleep(2)
            continue
        if response.status_code != 200:
            print(fclass="hl-string">'Error fetching {url}: {response.status_code}')
            break
        data = response.json()
        if class="hl-string">'totalcount' in data and data[class="hl-string">'totalcount'] == 0:
            break
        class=class="hl-string">"hl-comment"># Handle different response structures (key might be appointments, orders, etc.)
        keys = [k for k in data.keys() if k not in [class="hl-string">'totalcount', class="hl-string">'next', class="hl-string">'previous']]
        if keys:
            results.extend(data[keys[0]])
        
        if class="hl-string">'next' in data and data[class="hl-string">'next']:
             class=class="hl-string">"hl-comment"># Athena pagination often uses offset or a link
             class=class="hl-string">"hl-comment"># Adjust based on specific endpoint logic if next is not a full URL
             params[class="hl-string">'offset'] = params.get(class="hl-string">'offset', 0) + len(data[keys[0]])
        else:
            break
    return results

def extract_events():
    token = get_access_token()
    headers = {class="hl-string">'Authorization': fclass="hl-string">'Bearer {token}'}
    events = []

    class=class="hl-string">"hl-comment"># 1. Fetch Appointments (Patient Episodes)
    class=class="hl-string">"hl-comment"># Using /appointments/booked for the range
    print(class="hl-string">'Fetching appointments...')
    appt_params = {
        class="hl-string">'startdate': START_DATE,
        class="hl-string">'enddate': END_DATE,
        class="hl-string">'showclaimdetail': class="hl-string">'true',  class=class="hl-string">"hl-comment"># optimize to get claim info early
        class="hl-string">'ignorepartners': class="hl-string">'true'
    }
    appointments = fetch_paginated(fclass="hl-string">'{BASE_URL}/{PRACTICE_ID}/appointments/booked', headers, appt_params)

    for appt in appointments:
        episode_id = str(appt.get(class="hl-string">'appointmentid'))
        patient_id = appt.get(class="hl-string">'patientid')
        dept_id = appt.get(class="hl-string">'departmentid')
        provider_id = appt.get(class="hl-string">'providerid')
        
        class=class="hl-string">"hl-comment"># Base Activity: Appointment Scheduled
        if appt.get(class="hl-string">'date'): class=class="hl-string">"hl-comment"># capturing creation/scheduled date requires parsing history or booking info
            class=class="hl-string">"hl-comment"># Fallback to date if created date unavailable, but typically we want class="hl-string">'scheduleddatetime'
            sched_time = appt.get(class="hl-string">'scheduleddatetime', appt.get(class="hl-string">'date') + class="hl-string">' ' + appt.get(class="hl-string">'starttime'))
            events.append({
                class="hl-string">'PatientEpisodeId': episode_id,
                class="hl-string">'ActivityName': class="hl-string">'Appointment Scheduled',
                class="hl-string">'EventTimestamp': sched_time,
                class="hl-string">'SourceSystem': class="hl-string">'athenaOne',
                class="hl-string">'PatientId': patient_id,
                class="hl-string">'DepartmentName': dept_id
            })

        class=class="hl-string">"hl-comment"># Activity: Patient Checked In
        if appt.get(class="hl-string">'startcheckin'):
            events.append({
                class="hl-string">'PatientEpisodeId': episode_id,
                class="hl-string">'ActivityName': class="hl-string">'Patient Checked In',
                class="hl-string">'EventTimestamp': appt.get(class="hl-string">'date') + class="hl-string">' ' + appt.get(class="hl-string">'startcheckin'), class=class="hl-string">"hl-comment"># combining date and time
                class="hl-string">'SourceSystem': class="hl-string">'athenaOne',
                class="hl-string">'PatientId': patient_id
            })
            
        class=class="hl-string">"hl-comment"># Activity: Patient Discharged (Ambulatory Checkout)
        if appt.get(class="hl-string">'stopcheckin'):
            events.append({
                class="hl-string">'PatientEpisodeId': episode_id,
                class="hl-string">'ActivityName': class="hl-string">'Patient Discharged',
                class="hl-string">'EventTimestamp': appt.get(class="hl-string">'date') + class="hl-string">' ' + appt.get(class="hl-string">'stopcheckin'),
                class="hl-string">'SourceSystem': class="hl-string">'athenaOne',
                class="hl-string">'PatientId': patient_id,
                class="hl-string">'DischargeDisposition': class="hl-string">'Home' class=class="hl-string">"hl-comment"># Default for ambulatory
            })

        class=class="hl-string">"hl-comment"># Activity: Claim Submitted
        if class="hl-string">'claimid' in appt:
             class=class="hl-string">"hl-comment"># Ideally query /claims/{claimId} for exact submission timestamp
             class=class="hl-string">"hl-comment"># Using appointment date as proxy if detailed claim fetch skipped
             class=class="hl-string">"hl-comment"># Here we assume we fetch claim details:
             claim_res = requests.get(fclass="hl-string">'{BASE_URL}/{PRACTICE_ID}/claims/{appt["claimid"]}', headers=headers)
             if claim_res.status_code == 200:
                 claim_data = claim_res.json()
                 if claim_data.get(class="hl-string">'billedservicedate'):
                     events.append({
                        class="hl-string">'PatientEpisodeId': episode_id,
                        class="hl-string">'ActivityName': class="hl-string">'Claim Submitted',
                        class="hl-string">'EventTimestamp': claim_data.get(class="hl-string">'createddate', claim_data.get(class="hl-string">'billedservicedate')) + class="hl-string">' 12:00:00',
                        class="hl-string">'SourceSystem': class="hl-string">'athenaCollector',
                        class="hl-string">'PatientId': patient_id
                     })

        class=class="hl-string">"hl-comment"># Clinical Data Extraction (Requires Encounter Context)
        encounter_id = appt.get(class="hl-string">'encounterid')
        if encounter_id:
            class=class="hl-string">"hl-comment"># Activity: Initial Assessment (Vitals)
            vitals = fetch_paginated(fclass="hl-string">'{BASE_URL}/{PRACTICE_ID}/chart/{encounter_id}/vitals', headers, {})
            if vitals:
                class=class="hl-string">"hl-comment"># Take the first vital sign entry time
                first_vital = sorted(vitals, key=lambda x: x.get(class="hl-string">'date') or class="hl-string">'')[0]
                events.append({
                    class="hl-string">'PatientEpisodeId': episode_id,
                    class="hl-string">'ActivityName': class="hl-string">'Initial Assessment Completed',
                    class="hl-string">'EventTimestamp': first_vital.get(class="hl-string">'date') + class="hl-string">' ' + (first_vital.get(class="hl-string">'time') or class="hl-string">'00:00'),
                    class="hl-string">'SourceSystem': class="hl-string">'athenaClinicals',
                    class="hl-string">'PatientId': patient_id
                })

            class=class="hl-string">"hl-comment"># Activity: Diagnoses
            diagnoses = fetch_paginated(fclass="hl-string">'{BASE_URL}/{PRACTICE_ID}/chart/{encounter_id}/diagnoses', headers, {})
            for diag in diagnoses:
                events.append({
                    class="hl-string">'PatientEpisodeId': episode_id,
                    class="hl-string">'ActivityName': class="hl-string">'Diagnosis Confirmed',
                    class="hl-string">'EventTimestamp': diag.get(class="hl-string">'lastmodifieddatetime', appt.get(class="hl-string">'date') + class="hl-string">' 12:00:00'),
                    class="hl-string">'SourceSystem': class="hl-string">'athenaClinicals',
                    class="hl-string">'PrimaryDiagnosisCode': diag.get(class="hl-string">'diagnosiscode')
                })

            class=class="hl-string">"hl-comment"># Activity: Orders (Labs, Procedures, Discharge)
            orders = fetch_paginated(fclass="hl-string">'{BASE_URL}/{PRACTICE_ID}/chart/{encounter_id}/orders', headers, {})
            for order in orders:
                order_type = order.get(class="hl-string">'documentationtype', class="hl-string">'')
                created_time = order.get(class="hl-string">'createddatetime', appt.get(class="hl-string">'date') + class="hl-string">' 12:00:00')
                
                if class="hl-string">'LAB' in order_type.upper() or class="hl-string">'IMAGING' in order_type.upper():
                    events.append({
                        class="hl-string">'PatientEpisodeId': episode_id,
                        class="hl-string">'ActivityName': class="hl-string">'Diagnostic Test Ordered',
                        class="hl-string">'EventTimestamp': created_time,
                        class="hl-string">'SourceSystem': class="hl-string">'athenaClinicals'
                    })
                    class=class="hl-string">"hl-comment"># Activity: Specimen Collected (Inferred from status or specific fields if available)
                    if order.get(class="hl-string">'status') in [class="hl-string">'COMPLETED', class="hl-string">'REVIEW']:
                         events.append({
                            class="hl-string">'PatientEpisodeId': episode_id,
                            class="hl-string">'ActivityName': class="hl-string">'Specimen Collected',
                            class="hl-string">'EventTimestamp': order.get(class="hl-string">'lastmodifieddatetime', created_time), class=class="hl-string">"hl-comment"># Approximation
                            class="hl-string">'SourceSystem': class="hl-string">'athenaClinicals'
                        })
                
                if class="hl-string">'SURGERY' in order_type.upper() or class="hl-string">'PROCEDURE' in order_type.upper():
                     events.append({
                        class="hl-string">'PatientEpisodeId': episode_id,
                        class="hl-string">'ActivityName': class="hl-string">'Procedure Performed',
                        class="hl-string">'EventTimestamp': order.get(class="hl-string">'performdate', created_time),
                        class="hl-string">'SourceSystem': class="hl-string">'athenaClinicals'
                    })

                if class="hl-string">'DISCHARGE' in order_type.upper():
                    events.append({
                        class="hl-string">'PatientEpisodeId': episode_id,
                        class="hl-string">'ActivityName': class="hl-string">'Discharge Order Written',
                        class="hl-string">'EventTimestamp': created_time,
                        class="hl-string">'SourceSystem': class="hl-string">'athenaClinicals'
                    })

            class=class="hl-string">"hl-comment"># Activity: Test Results Received
            results = fetch_paginated(fclass="hl-string">'{BASE_URL}/{PRACTICE_ID}/chart/{encounter_id}/analyteresults', headers, {})
            for res in results:
                events.append({
                    class="hl-string">'PatientEpisodeId': episode_id,
                    class="hl-string">'ActivityName': class="hl-string">'Test Results Received',
                    class="hl-string">'EventTimestamp': res.get(class="hl-string">'analytedatetime', appt.get(class="hl-string">'date') + class="hl-string">' 12:00:00'),
                    class="hl-string">'SourceSystem': class="hl-string">'athenaClinicals'
                })

            class=class="hl-string">"hl-comment"># Activity: Treatment Plan
            class=class="hl-string">"hl-comment"># Check for Care Plan document or assessment note
            notes = fetch_paginated(fclass="hl-string">'{BASE_URL}/{PRACTICE_ID}/chart/{encounter_id}/progressnotes', headers, {})
            if notes:
                 events.append({
                    class="hl-string">'PatientEpisodeId': episode_id,
                    class="hl-string">'ActivityName': class="hl-string">'Treatment Plan Developed',
                    class="hl-string">'EventTimestamp': notes[0].get(class="hl-string">'lastmodifieddatetime', appt.get(class="hl-string">'date') + class="hl-string">' 12:00:00'),
                    class="hl-string">'SourceSystem': class="hl-string">'athenaClinicals'
                })

            class=class="hl-string">"hl-comment"># Activity: Medication Administered
            class=class="hl-string">"hl-comment"># Look for administration records
            meds = fetch_paginated(fclass="hl-string">'{BASE_URL}/{PRACTICE_ID}/chart/{encounter_id}/medications', headers, {})
            for med in meds:
                 if med.get(class="hl-string">'events'): class=class="hl-string">"hl-comment"># Check if administration events exist
                     for event in med[class="hl-string">'events']:
                         if event.get(class="hl-string">'type') == class="hl-string">'ADMINISTERED':
                             events.append({
                                class="hl-string">'PatientEpisodeId': episode_id,
                                class="hl-string">'ActivityName': class="hl-string">'Medication Administered',
                                class="hl-string">'EventTimestamp': event.get(class="hl-string">'eventdate'),
                                class="hl-string">'SourceSystem': class="hl-string">'athenaClinicals'
                             })

            class=class="hl-string">"hl-comment"># Activity: Patient Transferred (Inferred from Hospital Stay/ADT data if available)
            class=class="hl-string">"hl-comment"># Mocking logic: If department ID changes in encounter history (requires fetching history)
            class=class="hl-string">"hl-comment"># For standard API, we check if class="hl-string">'admissionstatus' indicates transfer
            pass class=class="hl-string">"hl-comment"># Complexity high for standard extraction without Hospital scope

            class=class="hl-string">"hl-comment"># Activity: Follow-Up Appointment Scheduled
            class=class="hl-string">"hl-comment"># We check if this patient has a FUTURE appointment created around this time
            class=class="hl-string">"hl-comment"># This requires a separate query or processing the full appointment list post-loop
            class=class="hl-string">"hl-comment"># Simplified: Logic placeholder
            pass 

    class=class="hl-string">"hl-comment"># Post-processing for Follow-Ups (simplified)
    class=class="hl-string">"hl-comment"># Sort all appts by patient and date, check next appt creation date
    
    class=class="hl-string">"hl-comment"># Convert to DataFrame
    df = pd.DataFrame(events)
    df[class="hl-string">'LastDataUpdate'] = datetime.now().isoformat()
    
    class=class="hl-string">"hl-comment"># Final Validation and Export
    if not df.empty:
        df[class="hl-string">'EventTimestamp'] = pd.to_datetime(df[class="hl-string">'EventTimestamp'])
        df.sort_values(by=[class="hl-string">'PatientEpisodeId', class="hl-string">'EventTimestamp'], inplace=True)
        df.to_csv(OUTPUT_FILE, index=False)
        print(fclass="hl-string">'Successfully exported {len(df)} rows to {OUTPUT_FILE}')
    else:
        print(class="hl-string">'No events extracted.')

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

您的患者旅程数据模板

您的患者旅程数据模板

患者旅程属性

患者旅程活动

提取指南

athenaNet 报表生成器自定义 CSV 导出

步骤

配置

a 查询示例 config

使用 Python 进行 athenaOne API 事件提取

步骤

配置

a 查询示例 python

准备好开始了吗？

立即优化您的患者旅程，消除瓶颈

您的患者旅程数据模板

您的患者旅程数据模板

患者旅程属性

患者旅程活动

提取指南

athenaNet 报表生成器自定义 CSV 导出

步骤

配置

a 查询示例 config

使用 Python 进行 athenaOne API 事件提取

步骤

配置

a 查询示例 python

我们重视您的隐私

准备好开始了吗？

立即优化您的患者旅程，消除瓶颈