您的患者旅程数据模板

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岁及以上。
说明将患者划分为不同人口统计群体，是“年龄组与诊断旅程对比”仪表板的必要字段。有助于识别流程低效或结果是否对特定年龄群体产生不成比例的影响，例如老年患者或儿科患者。为什么重要用于医疗流程分析的标准人口统计分组。获取位置根据患者出生日期和StartTime推导。示例 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	订单类别，例如实验室、影像或处方。
说明对就诊事件期间下达的临床订单进行分类，是“诊断检测周期时间”仪表板的关键属性。分析人员可以分别衡量实验室检测和影像检查的周期时间，因为两者通常具有不同的服务级别协议和瓶颈。为什么重要按具体类型划分诊断流程，以便分析周期时间。获取位置 athenahealth中的“ordertype”字段，或订单API中的“class”字段。示例实验室检查影像检查处方医疗操作
	诊断结果状态 DiagnosticResultStatus	诊断医嘱的结果状态（例如阳性、正常）。
说明记录检测结果的总体状态，为“诊断检测前置时间”和后续治疗决策提供背景信息。用于分析异常结果是否会比正常结果更快触发后续行动。为什么重要将顺序流与临床结果关联起来。获取位置 athenahealth实验室结果观测字段。示例正常异常危急

必需建议可选

患者就诊旅程活动

以下是需要记录在事件日志中的关键流程步骤和里程碑，用于准确发现流程并优化患者体验。

7 建议 8 可选

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

建议可选

提取指南

如何从athenahealth获取数据

步骤

使用安全凭据登录athenaNet门户。
通过主菜单进入Reports模块，通常路径为“Reports”>“Report Builder”或“Custom Reports”。
点击“Create New Report”，并选择“Encounters”或“Patient Visit”作为主要报告类别。这样可以确保明细数据围绕具体患者就诊过程展开。
在“Columns”或“Fields”选择区域中，搜索并添加Patient Identifier（Patient ID）和Encounter Identifier（Encounter ID），将其作为Case Identifier。
添加属性所需的人口统计和上下文列，例如Department、Provider Name、Patient Age和Primary Diagnosis Code。
添加与各项活动对应的具体时间戳列。您可能需要在不同字段组中查找，例如在“Scheduling”中查找Appointment Scheduled，在“Check-in”中查找Check-in Time，在“Billing”中查找Claim Submitted。可选时，请选择“Date and Time”格式，而不只是Date。
为报告配置筛选条件。将“Encounter Date”范围设为所需分析窗口，例如Last 6 Months，并排除“Cancelled”或“No Show”预约，以聚焦已完成的就诊过程。
点击“Run Report”预览数据。确认各列包含数据，而不是全部为空值。
选择“Export”，并将“CSV”（Comma Separated Values）作为格式。请勿使用Excel/XLSX格式化功能，以免增加后续处理复杂度。
将文件保存到本地，文件名为“athena_raw_export.csv”。
由于报告以每次就诊一行、多个日期列的宽表格式导出，您必须将其转换为事件日志的长表格式。请使用Query部分提供的Python脚本完成转换。
运行脚本生成最终文件“process_mining_event_log.csv”，并将该文件上传到ProcessMind。

配置

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

a 示例查询 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 Client Credentials授权流程，获取访问令牌。后续所有API调用都必须在Authorization请求头中携带该令牌，格式为Bearer token。
定义范围：设置Practice ID和数据提取日期范围。Athenahealth API按诊所区分，因此如果您管理多个诊所，必须遍历各个Practice ID。
提取预约：查询/appointments/booked端点或/appointments/changed端点，获取基础就诊记录。将预约ID映射到PatientEpisodeId。根据创建日期提取“Appointment Scheduled”，根据签到时间提取“Patient Checked In”。
遍历就诊记录以获取临床事件：针对每个预约ID（就诊记录）查询临床子端点。调用/chart/{encounterId}/vitals获取“Initial Assessment Completed”；调用/chart/{encounterId}/orders获取“Diagnostic Test Ordered”、“Specimen Collected”（根据状态判断）和“Discharge Order Written”。
获取结果和诊断：查询/chart/{encounterId}/analyteresults或/chart/{encounterId}/labresults获取“Test Results Received”。查询/chart/{encounterId}/diagnoses获取“Diagnosis Confirmed”。
提取治疗和操作：查询/chart/{encounterId}/plan获取“Treatment Plan Developed”。查询/chart/{encounterId}/procedures获取“Procedure Performed”。查询/chart/{encounterId}/medications/admin获取“Medication Administered”。
识别ADT和索赔事件：使用预约状态历史或特定医院端点（如可用）捕获“Patient Transferred”和“Patient Discharged”。查询与预约相关的/claims端点，提取“Claim Submitted”。
关联后续预约：在就诊时间范围内，搜索同一患者ID后续创建的预约，识别“Follow-Up Appointment Scheduled”。
转换并扁平化：将嵌套JSON响应解析为事件字典列表。将所有时间戳转换为ISO 8601格式，并将字段映射到ActivityName、PatientEpisodeId和EventTimestamp。
导出数据：将字典列表加载到pandas DataFrame中，完成最终清理（处理空值），并导出为可供ProcessMind导入的CSV文件。

配置

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

准备好开始了吗？

开启优化患者医疗服务的旅程。此模板是您的第一步，帮助您发掘有价值的洞察，改善医疗服务结果。

返回改进指南下载模板

立即优化患者就诊流程，消除瓶颈

精准定位athenahealth中的低效环节，将患者就诊流程周期时间缩短30%。

开始免费试用

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

您的患者旅程数据模板

您的患者旅程数据模板

患者就诊旅程属性

患者就诊旅程活动

提取指南

athenaNet Report Builder自定义CSV导出

步骤

配置

a 示例查询 python

使用Python从athenaOne API提取事件

步骤

配置

a 示例查询 python

准备好开始了吗？

立即优化患者就诊流程，消除瓶颈

您的患者旅程数据模板

您的患者旅程数据模板

患者就诊旅程属性

患者就诊旅程活动

提取指南

athenaNet Report Builder自定义CSV导出

步骤

配置

a 示例查询 python

使用Python从athenaOne API提取事件

步骤

配置

a 示例查询 python

准备好开始了吗？

立即优化患者就诊流程，消除瓶颈

我们重视您的隐私