患者就医旅程数据模板

athenahealth

← 返回改进指南

患者就医旅程属性(19) 患者就医旅程活动(15) 数据提取指南(2)

患者就医旅程数据模板

此模板介绍分析患者就医旅程所需数据的收集方法，列出需要跟踪的关键属性和活动，并说明如何从athenahealth中提取这些信息。借助此资源，您可以为有效开展流程挖掘做好数据准备。

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

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

下载模板

患者就医旅程属性

建议将这些数据字段纳入事件日志，以便全面分析并了解患者就医旅程。

5 必填 9 建议填写 5 选填

	名称	说明
	事件时间戳 EventTimestamp	活动发生的具体日期和时间。
说明记录活动发生的准确时间，用于按时间顺序排列事件，并计算各步骤之间的时长。这是开展时间分析的关键数据，包括周期时间、等待时间和吞吐量分析。建议使用高精度时间戳，以确定同一天发生的多个事件的先后顺序。重要性用于排列事件顺序并计算所有基于时间的KPI。获取途径 athenahealth表中与状态变更或创建日期相关的时间戳字段。示例 2023-10-12T08:30:00Z2023-10-12T09:15:22Z2023-10-15T14:20:00Z
	患者就诊周期 PatientEpisodeId	用于标识特定患者就诊周期或诊疗历程的唯一标识符。
说明此属性作为核心案例标识符，将患者某一特定诊疗阶段或病情相关的所有活动归为一组。它将预约、诊断医嘱和出院手续等分散事件串联成完整的诊疗历程。在分析中，此ID是流程挖掘的主键，可用于还原端到端流程。它还能确保同一患者因不同病情多次就诊时，被视为不同的流程实例。重要性对于界定分析中单个流程实例的范围至关重要。获取途径通过对Encounter ID分组得出，或关联到athenahealth中的特定Episode of Care ID。示例 EP-2023-88491EP-2023-99102ENC-55412-GRP
	数据最近更新时间 LastDataUpdate	数据最近一次提取或刷新的时间戳。
说明用于说明分析所用数据的新鲜度，帮助用户判断查看的是实时数据还是历史快照。该属性还可用于管理数据管道，确保仪表板反映流程的最新状态。重要性对于数据治理以及用户对仪表板数据时效性的信任至关重要。获取途径 ETL执行时的系统时间。示例 2023-11-01T12:00:00Z2023-11-02T06:00:00Z
	活动名称 ActivityName	患者诊疗历程中发生的事件或执行的任务名称。
说明用于标识流程中发生的具体步骤，例如“患者已登记”“已开具诊断检查医嘱”或“已施用药物”。此文本用于定义流程图中的节点。该属性用于呈现流程并识别操作顺序。统一这些名称对于生成清晰易读、无聚类的流程图至关重要。重要性用于定义流程图中的步骤，是开展流程挖掘的必备属性。获取途径可从审计日志、预约状态变更或理赔明细描述中获取。示例预约已安排患者已签到诊断检查已开具患者已出院
	源系统 SourceSystem	数据来源系统的名称。
说明用于标识生成记录的IT系统，本例中为“athenahealth”。在数据可能经过整合的多系统环境中，这一属性尤其有用。分析人员可按数据源筛选视图，并排查特定系统的数据质量问题。重要性为多系统流程挖掘提供数据沿袭信息和背景。获取途径硬编码值或系统配置ID。示例 athenahealthAthenaOneAthenaPractice
	主要诊断代码 PrimaryDiagnosisCode	与就诊周期相关的主要ICD-10代码。
说明用于对患者诊疗历程的临床原因进行分类，为“年龄组与诊断历程对比”提供背景信息。分析人员可按病情（如肺炎或骨折）划分诊疗历程，因为不同病情的预期路径和周期时间可能差异很大。重要性便于比较同类案例；不同诊断的周期时间可能相差很大。获取途径就诊或理赔中的诊断字段（ICD-10）。示例 J18.9I10E11.9
	事件结束时间 EventEndTime	特定活动完成的时间。
说明记录活动完成时间，以便计算该步骤本身的实际处理时长，并与活动开始前的等待时间区分开来。该属性可用于分析资源效率，并识别人工执行时间超出预期的任务。重要性可用于计算实际处理时间与等待时间。获取途径 athenahealth中的结账时间、结果核实时间或特定完成时间戳。示例 2023-10-12T09:00:00Z2023-10-12T10:45:00Z
	出院去向 DischargeDisposition	患者出院时的去向或状态。
说明用于说明患者就诊周期结束后的去向，例如回家、转入专业护理机构或临终关怀机构，是分析“出院计划与再入院趋势”的重要属性。该属性可反映出院计划的复杂程度，并帮助评估照护衔接的效果。重要性对于制定出院计划和评估再入院风险至关重要。获取途径 athenahealth就诊字段或医院出院记录。示例回家专业护理机构转入短期住院医院死亡
	医疗服务提供者姓名 ProviderName	执行活动的医疗专业人员姓名。
说明用于识别负责该事件的具体医生、护士或技术人员，是资源利用率分析的核心属性。您可以比较不同员工的吞吐量、周期时间等绩效指标，以发现培训需求或工作量失衡。重要性用于分析“资源利用率”和工作交接的关键属性。获取途径将athenahealth中的“providerid”解析为医疗服务提供者目录中的姓名。示例 Smith医生Jones护士Adams技术人员
	就诊类型 EncounterType	就诊类别，例如门诊、远程医疗或急诊。
说明用于定义所提供医疗服务的方式或场景，在通用数据模型中相当于“案例类型”。不同就诊类型的预期流程和账单要求各不相同。按此属性筛选，有助于避免在周期时间分析中比较不可比的案例。重要性用于区分远程医疗与现场就诊等不同流程变体。获取途径 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”解析为科室名称。示例急诊科内科放射科
	医嘱类型 OrderType	医嘱类别，例如检验、影像检查或处方。
说明用于对就诊周期内开具的临床医嘱进行分类，是“诊断检查前置时间”仪表板的关键属性。分析人员可分别衡量检验和影像检查的前置时间，因为两者的服务级别协议和瓶颈往往不同。重要性用于划分诊断流程，以便针对性分析前置时间。获取途径 athenahealth中的“ordertype”字段，或订单API中的“class”字段。示例检验影像检查处方操作
	是否返工 IsRework	用于标记该活动是否重复发生。
说明布尔标记；如果同一案例中的活动发生次数超过一次，则值为true。该属性支持“活动频次与返工循环”仪表板。您可以快速筛选出包含返工的案例，便于计算“返工活动频次”KPI。重要性用于识别流程效率问题和冗余步骤。获取途径 ETL期间根据每个CaseId下的活动发生次数计算得出。示例 truefalse
	理赔状态 ClaimStatus	与诊疗相关的财务理赔状态。
说明用于说明账单理赔的当前状态，例如“已提交”“已拒绝”或“已付款”。该属性与“理赔已提交”活动及财务吞吐量相关。有助于识别临床文档问题是否导致后端财务处理延误。重要性将临床效率与收入周期表现关联起来。获取途径 athenahealth中的“claimstatus”字段。示例已计费暂缓已提交
	诊断结果状态 DiagnosticResultStatus	诊断医嘱的结果状态，例如阳性或正常。
说明记录检查结果的总体状态，为“诊断检查前置时间”分析和后续治疗决策提供背景。该属性可用于分析异常结果是否比正常结果更快触发后续行动。重要性将流程与临床结果关联起来。获取途径 athenahealth检验结果观察字段。示例正常异常严重
	费用总额 TotalChargeAmount	针对该活动或就诊周期收取的金额。
说明与活动相关的金额或理赔总额，可用于开展基于成本的流程挖掘和财务影响分析。该属性可用于识别治疗路径中的高成本变体，并分析流程效率与财务结果之间的关系。重要性为流程分析增加财务维度。获取途径 athenahealth理赔或费用表中的“amount”或“totalcharge”字段。示例 150.002500.5045.00

必填建议填写选填

患者就医旅程活动

建议在事件日志中记录这些关键流程步骤和里程碑，以准确发现并优化患者就医体验。

7 建议填写 8 选填

	活动	说明
	出院医嘱已开具	医生或获授权的医疗服务提供方已开具正式医嘱，安排患者出院。此事件在EHR的CPOE模块中创建，并带有时间戳。
重要性此活动启动出院流程。从开具医嘱到实际出院的时间，是衡量“出院计划周期时间”的关键绩效指标。获取途径此事件记录在医嘱表中，可通过特定的“出院”医嘱类型及其创建时间戳识别。采集医嘱表中会创建一条医嘱类型为“出院”的新记录。事件类型 explicit
	初步评估已完成	表示首次临床评估已完成，例如分诊或护理评估，此时会记录生命体征和主要症状。通常可根据首次签署临床记录的时间戳，或就诊评估表单的完成时间推断此事件。
重要性此里程碑标志着临床护理开始。从签到到此活动的时长，是衡量患者初始等待时间和资源响应速度的重要指标。获取途径可根据athenaClinicals中特定临床文档或流程表的创建或签署时间戳推断。需要确定与分诊或接诊相关的文档类型。采集查找该次就诊的临床记录、生命体征流程表或特定接诊表单中的最早时间戳。事件类型 inferred
	患者已出院	患者已正式出院，院内就医流程结束。这是住院就诊的最后一条ADT事件，并带有准确的时间戳。
重要性此事件标志着患者主要就医流程的结束，是衡量整体“患者就医周期时间”的终点，也是再入院分析的重要依据。获取途径此事件明确记录在ADT系统或患者就诊表中，将就诊最终状态标记为“已出院”，并附有时间戳。采集患者就诊状态更新为“已出院”，并记录一条ADT事件。事件类型 explicit
	患者已签到	表示患者已到达，并为预约或就诊办理正式签到。通常会在athenaClinicals或athenaCommunicator的预约记录中明确记录状态变更。
重要性这是患者现场就医流程的明确起点，也是衡量等待时间和临床就诊整体周期时间的重要依据。获取途径此事件记录为预约表或就诊表中的状态更新。请查找“已签到”状态及其对应的时间戳。采集预约或就诊对象的状态变更会连同时间戳一起记录。事件类型 explicit
	操作已完成	已为患者实施临床操作，例如手术或专科治疗。此事件会明确记录在临床文档中，操作记录通常包含具体的开始和结束时间。
重要性操作是患者治疗过程中的重要里程碑。分析操作前后的活动，有助于优化术前和术后工作流。获取途径此事件记录在athenaClinicals的操作记录或特定临床流程表中。事件时间戳根据记录的操作开始或结束时间确定。采集系统会创建操作记录或日志，并标注操作发生的时间戳。事件类型 explicit
	检查结果已收到	诊断检查结果已最终确认，并可在患者病历中查看。通常在实验室或影像系统将结果发送回athenahealth时记录，并生成带有时间戳的条目。
重要性收到结果是后续临床决策的重要触发点，例如确诊和制定治疗计划。此事件是衡量诊断周转时间的终点。获取途径此事件位于结果表或诊断表中，并关联原始医嘱。患者病历中登记或收到结果时的时间戳即为事件时间。采集系统会创建一条带有时间戳的新结果记录，通常通过LIS或RIS接口传入。事件类型 explicit
	诊断已确认	临床医生正式为患者当前就诊的病情指定或确认诊断。可根据与患者就诊关联的主要诊断代码（如ICD-10）的创建时间或“最后更新”时间推断此事件。
重要性这是决定后续治疗路径的重要里程碑。分析此后活动的差异，有助于了解并规范护理方案。获取途径可根据患者问题列表或就诊诊断数据推断。诊断代码发生变更或最终确认时，其对应的时间戳可作为此事件的标志。采集识别该次就诊主要诊断的添加或更新时间戳。事件类型 inferred
	患者已转运	患者已从一个护理单元或部门转至另一个，例如从急诊科转入住院病房。此事件会通过电子健康记录（EHR）中的入院、出院和转运（ADT）记录明确捕获。
重要性此活动对于分析部门间交接和院内患者流转至关重要，有助于发现“患者交接时间”及资源分配方面的瓶颈。获取途径此事件记录在ADT事件日志或患者跟踪表中。每条转运记录都包含患者、转出和转入地点以及时间戳。采集患者转运时，系统会生成带有时间戳的ADT消息或事件日志条目。事件类型 explicit
	样本已采集	表示已从患者处采集血液或尿液等生物样本，用于实验室检查。通常会在实验室模块中明确记录此事件，或通过医嘱状态更新记录。
重要性这是诊断检查流程中的关键里程碑。分析从开具检查到采集样本、再到结果出具的时间，有助于发现实验室工作流中的重要瓶颈。获取途径此事件可记录为实验室医嘱的状态变更，也可能作为独立事件记录在与athenahealth对接的实验室信息系统中。请查找“已采集”状态及其时间戳。采集此事件记录为实验室医嘱状态更新，或记录在专用样本跟踪模块中。事件类型 explicit
	治疗计划已制定	表示临床医生已正式制定并记录患者的治疗计划。可根据特定“护理计划”文档的创建或签署时间，或相关治疗医嘱集的创建时间记录此事件。
重要性此活动正式确定预期的临床路径，是衡量方案执行一致性和分析护理差异的重要节点。获取途径此事件可能记录在临床文档表或医嘱表中，对应已签署护理计划记录的时间戳，或一组协调治疗医嘱的提交时间。采集此事件的时间戳为特定治疗计划文档或医嘱集的创建或最终确认时间。事件类型 explicit
	理赔申请已提交	系统已生成患者就诊期间所提供服务的理赔申请，并提交给付款方。此事件明确记录在athenaCollector收入周期管理模块中。
重要性虽然属于行政环节，但这项活动对于分析与临床就诊并行的收入周期流程至关重要，有助于发现临床诊疗与账单处理之间的延误。获取途径可在理赔或账单表中找到。事件以理赔创建或提交时间戳为标记。采集创建理赔记录，并将状态更新为“已提交”，同时记录时间戳。事件类型 explicit
	药物已给药	临床工作人员已向患者实际给药。athenaClinicals的用药记录（MAR）模块会明确记录此事件，并为每次给药标注准确的时间戳。
重要性此活动代表直接治疗干预。分析其时间有助于衡量“首次治疗时间”，并确保按计划给药。获取途径此事件记录在MAR数据表中。每条给药记录都包含患者ID、药物ID、剂量和给药时间戳。采集每次记录已给药时，MAR中都会生成一条带有时间戳的记录。事件类型 explicit
	诊断检查已开具	医疗服务提供方开具诊断检查，例如实验室检查、影像扫描或其他操作。这是通过athenaClinicals中的CPOE（计算机化医嘱录入）功能创建的明确事件，并带有时间戳。
重要性这是启动诊断子流程的关键决策点。跟踪此活动对于分析从开具检查到结果出具的“诊断检查周期时间”指标至关重要。获取途径此事件记录在医嘱表中。每条医嘱都包含患者标识符、医嘱名称、医嘱状态和创建时间戳。采集系统医嘱表中会创建一条带有时间戳的新记录。事件类型 explicit
	随访预约已安排	患者完成主要治疗或出院后，已安排随访预约。用户在athenaCommunicator排期模块中创建新预约时，系统会明确记录此事件。
重要性此活动有助于了解出院后护理协作及其对再入院率等结局的影响，也体现了护理的连续性。获取途径此事件记录在预约表中，可通过出院日期之后的预约创建时间戳识别。采集排期系统中会创建一条新的预约记录。事件类型 explicit
	预约已安排	表示已为患者预约。用户在athenahealth排期模块athenaCommunicator中创建并确认新预约时，系统会明确记录此事件。
重要性这是许多患者就医流程的起点。分析从预约到签到的时间，有助于了解患者就医可及性和就诊前流程效率。获取途径这是预约或排期表中明确记录的事件，通常包含创建时间戳和患者ID。采集在排期模块中创建预约记录时，系统会记录此事件。事件类型 explicit

建议填写选填

数据提取指南

如何从athenahealth获取数据

步骤

使用您的安全凭据登录athenaNet门户。
从主菜单进入Reports模块，通常位于“Reports”>“Report Builder”或“Custom Reports”下。
点击“Create New Report”，并选择“Encounters”或“Patient Visit”作为主要报告类别。这样可以围绕具体患者就诊周期提取细粒度数据。
在“Columns”或“Fields”选择区域中，搜索并添加患者标识符（Patient ID）和就诊标识符（Encounter ID），将其用作案例标识符。
添加属性所需的人口统计和背景信息列，例如部门、医务人员姓名、患者年龄和主要诊断代码。
添加与各项活动对应的时间戳列。您可能需要在不同字段组中查找，例如在“Scheduling”中查找预约时间，在“Check-in”中查找签到时间，在“Billing”中查找理赔提交时间。请尽量选择“Date and Time”格式，而不只是日期。
为报告设置筛选条件。将“Encounter Date”范围设为所需的分析时段，例如最近6个月，并排除“Cancelled”或“No Show”预约，以便聚焦已完成的就医旅程。
点击“Run Report”预览数据。确认各列包含数据，而非全部为空值。
选择“Export”，并将格式设为“CSV”（逗号分隔值）。请勿使用Excel/XLSX格式功能，以免增加处理难度。
将文件保存到本地，命名为“athena_raw_export.csv”。
报告以宽表格式导出，每次就诊一行并包含多个日期列，因此需要将其转换为长表格式的事件日志。请使用查询部分提供的Python脚本完成转换。
运行脚本生成最终文件“process_mining_event_log.csv”，并将其上传至ProcessMind。

配置

报告类别：选择“Encounter Activity”或“Clinicals”作为基础类别，以涵盖大部分临床工作流步骤。
日期范围：首次提取建议限制在3至6个月内，以避免报告超时或行数超限。标准视图通常最多约50,000行。
列选择：确保选中明确的时间戳字段。请使用绝对时间字段“Check-in Date/Time”，而非“Days since check-in”等计算字段。
空值处理：将缺失数据设置为空白单元格，而非“N/A”等自定义文本，以便脚本处理。
部门筛选：如果组织规模较大，可按具体服务地点或部门筛选，例如“Cardiology”或“Orthopedics”，以缩小数据范围。
权限：用户需要在athenaNet中拥有“Report Creator”或“Clinical Reporting”权限，才能使用自定义报告生成器。

查询示例 python

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

class=class="hl-string">"hl-comment"># Configuration: Input and Output filenames
INPUT_FILE = class="hl-string">'athena_raw_export.csv'
OUTPUT_FILE = class="hl-string">'patient_journey_event_log.csv'

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

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

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

class=class="hl-string">"hl-comment"># Initialize list to hold event rows
event_log_data = []

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

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

class=class="hl-string">"hl-comment"># Create DataFrame from the list of events
event_log_df = pd.DataFrame(event_log_data)

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

class=class="hl-string">"hl-comment"># Remove rows where timestamp parsing failed
event_log_df = event_log_df.dropna(subset=[class="hl-string">'EventTimestamp'])

class=class="hl-string">"hl-comment"># Sort by Episode and Time
event_log_df = event_log_df.sort_values(by=[class="hl-string">'PatientEpisodeId', class="hl-string">'EventTimestamp'])

class=class="hl-string">"hl-comment"># Output to CSV
event_log_df.to_csv(OUTPUT_FILE, index=False)

print(fclass="hl-string">"Transformation complete. Processed {len(df)} encounters into {len(event_log_df)} events.")
print(fclass="hl-string">"File saved to: {OUTPUT_FILE}")

步骤

注册API应用：登录athenahealth Developer Portal并注册新应用，以获取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 Developer Portal获取有效的Client ID和Secret，并拥有生产环境访问权限。
Practice ID：医疗机构的唯一数字标识符。若要提取医疗网络的数据，必须针对每个诊所运行脚本。
日期范围：由于临床细粒度数据量较大，建议提取最近3至6个月的数据。
速率限制：athenaOne API有严格的速率限制，例如每秒5次调用。脚本包含休眠函数，但企业级大规模提取应实现指数退避机制。
范围：确保令牌具有athena/service/Athenanet.MDP.*范围，或具有预约、病历和财务数据对应的细分范围。
分页：大多数端点返回的记录数量有限，默认值为1000。脚本必须处理next或offset参数。
部门筛选：若要分析特定流程，例如急诊与门诊流程，请在获取预约数据时按部门ID筛选。

查询示例 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()

准备好开始了吗？

开启优化患者护理之旅。此模板将帮助您迈出第一步，发掘有价值的洞察并改善医疗服务成效。

返回改进指南

下载模板

立即优化患者就医旅程，消除瓶颈

找出athenahealth中的低效环节，将患者就医旅程周期时间缩短30%。

开始免费试用

无需信用卡。立即开始优化。

患者就医旅程数据模板

患者就医旅程数据模板

患者就医旅程属性

患者就医旅程活动

数据提取指南

athenaNet Report Builder自定义CSV导出

步骤

配置

查询示例 python

使用Python通过athenaOne API提取事件

步骤

配置

查询示例 python

准备好开始了吗？

立即优化患者就医旅程，消除瓶颈