Why trading.hyena17 runs past 09:30: request timeouts and slow service readiness

william 收录于 Linux

2026-11-12 2026-11-12 约 6300 字预计阅读 30 分钟

/2026-11-12-airflow-trading-hyena17-task-timing/morning-timeline.png

I wanted to know which tasks in trading.hyena17 make the morning setup run past 09:30. Sorting by duration was misleading. Some tasks deliberately run until 15:00, while a task that takes only five seconds can still start ninety seconds late.

The Nov 12 run has two main problems. Stock-weight preparation takes 352.6 seconds and leaves only 4.4 seconds before 09:30. Downstream, the slowest shared-service startup takes another 167.3 seconds and finishes at 09:35:15. Redis itself takes 4.8 seconds. Its late start comes from its dependencies.

Full morning timeline for every fund, account, and colo, with 09:30 marked in red

Open the full-size timeline to read all 252 task rows. The eight sections contain all 26 fund targets, 34 zfu_param generation tasks, 34 transfers to R7, 22 transfers to colos, 34 zfu_check validation tasks, 68 stock account/colo launch tasks for 34 accounts, 22 shared-service startups, and nine futures account launches, plus stock weight, Redis, and futures risk. This chart uses the first recorded morning attempts and includes both stock-market colos for each account.

What I measured

I read the production Airflow metadata and selected task logs without running or clearing any tasks. The snapshot starts at 17:08 UTC+8 on Nov 12, after the trading session.

Airflow 2.7.3, LocalExecutor, MySQL metadata, Asia/Shanghai timezone.
The latest 20 scheduled dates, Oct 16 through Nov 12, 2026. The DAG is scheduled at 09:05 on weekdays.
7,301 current task-instance rows and 15,113 task-state audit events. Manual DAG runs are outside the sample.
Historical duration comparisons use the 17 dates with an observed successful stock-weight run, excluding the known Nov 11 incident. Each task has its own sample count.

All displayed times are UTC+8. The naive timestamps read from MySQL represent UTC. I checked that conversion against task logs, where the 01:05 UTC scheduled run appears at 09:05 local time.

The report uses the original Airflow task IDs, fund codes, account names, and colocation names. Stock target position: ZX7 is zsun_pre_trade_stock_target_pos_ZX7: it requests pre-trade stock target positions for fund ZX7 and waits for completion. Shared services: Colo153 is h17.trading.shared_Colo153, the shared-service startup task on Colo153. Charts abbreviate the task family for readability; tables and CSV downloads retain the Airflow identifiers.

Reruns make the current task table an incomplete history

The current dag_run.start_date for Nov 12 says 16:44 local time. The same run contains tasks that started at 09:05. Someone reran part of the DAG in the afternoon, so the run-level timestamps do not describe the original morning execution.

I reconstructed attempts by pairing each task’s running audit event with its next success or failed event. A second running event without a terminal event leaves the earlier attempt unresolved. When a current task-instance or failure record matches an attempt, its more precise timestamps replace the audit timestamps. Original queue timestamps are used only when that matching task-instance record survives.

For morning duration comparisons, I select the first observed attempt per task between 09:00 and noon on the scheduled date. Failed and unresolved attempts stay in the data, but do not enter successful-runtime percentiles. Empty operators can have metadata-only records. Other metadata-only successes without evidence of a running attempt are excluded, because marking a task successful does not prove that its work ran.

Across the 20 dates, those first morning attempts contain 5,759 successes, 107 failures, and 93 unresolved outcomes. For Nov 12, all 361 observed first morning task or gate instances succeed. That does not mean the application made no retries inside those tasks.

The history chart uses the first successful morning completion per task, including a recovery attempt when necessary. Oct 16 and Oct 19 lack successful stock-weight records in this reconstruction. Gray cells mean missing evidence, not an on-time completion.

The delay begins before 09:30

Historical readiness times over 20 scheduled dates

Stock-weight runtime changes visibly after Oct 23. Its median is 108.8 seconds across Oct 20–23, then 341.2 seconds across the 13 later non-incident dates. This is an observational change, not a controlled benchmark. The sample contains 14 different DAG hashes, and I have not isolated a backend release, input-size change, or configuration change that explains it.

The Nov 12 task starts at 09:24:02.972 and succeeds at 09:29:55.607. The application log shows an accepted request, a timeout, a second request, and eventually completion:

1
2
3
4
5


09:24:04.033  sending request: stock_daily {'api': 'pre_trade', 'mode': 'stock_weight'}
09:24:04.056  response received: ... "status": "received" ...
09:29:04.065  retrying after request timeout: stock_daily {'api': 'pre_trade', 'mode': 'stock_weight'}
09:29:04.104  sending request: stock_daily {'api': 'pre_trade', 'mode': 'stock_weight'}
09:29:55.385  response received: ... "status": "completed" ...

These are shortened excerpts. The sanitized original lines preserve the timestamps and messages.

The first request is accepted promptly. Most of the task’s elapsed time is then spent awaiting application completion. The log does not establish whether the backend is computing, queued behind other requests, or failing to deliver a completion response. The 300-second timeout is an observed symptom, not 300 seconds of proven avoidable work.

The selected stock-weight task logs survive only for Nov 11 and Nov 12. Both contain this retry pattern. I cannot use those two logs to claim that the earlier slow dates had the same cause.

Which tasks are slow after the open

Today, 253 of 316 first-attempt setup or gate instances finish after 09:30. This excludes intentional wait tasks and the long-lived trading sessions. That is a deadline-miss count, not 253 separate root causes. Many short tasks simply inherit a late upstream completion.

The following times are from the original morning attempts, even where the current metadata was overwritten by a later rerun.

Task or family	Starts	Finishes	Runtime
`zsun_pre_trade_stock_weight`	09:24:02.972	09:29:55.607	352.6 s
First successful stock target, `zsun_pre_trade_stock_target_pos_ZX7`	09:29:58.460	09:30:02.918	4.5 s
Last successful stock target, `zsun_pre_trade_stock_target_pos_ZX14`	09:29:58.721	09:31:31.982	93.3 s
`redis_tpc`	09:31:34.769	09:31:39.523	4.8 s
`h17.trading.shared_Colo153`	09:32:27.945	09:35:15.291	167.3 s
`h17.trading.shared_Colo152`	09:32:27.890	09:35:10.548	162.7 s
`h17.trading.shared_Colo181`	09:32:17.767	09:34:47.845	150.1 s

The 26 stock-target tasks all start within 0.6 seconds of one another, after stock weight finishes. All 26 finish after 09:30. Their median runtime is 34.3 seconds, with a maximum of 93.3 seconds.

Four stock-target tasks, DC5, ZX14, ZX8, and ZZ1, retry after a 60-second request timeout. Their runtimes range from 80.5 to 93.3 seconds. Airflow records one attempt for each because these are retries inside the application client.

Tasks ranked by their own running time after 09:30

A stock target that runs from 09:29:58 to 09:31:32 contributes about 92 seconds to this chart. Its full runtime is about 93 seconds. These quantities differ from its completion delay relative to 09:30, and none measures CPU consumption.

Shared services spend most of their runtime waiting for readiness

For the three slowest shared-service tasks, I matched the log line where the readiness poll identifies the process log with the first Airflow log line that reports a worker thread:

Shared-service task	Observed readiness interval	Interval	Whole task
`Colo153`	09:32:33.797 → 09:35:12.466	158.7 s	167.3 s
`Colo152`	09:32:34.089 → 09:35:07.734	153.6 s	162.7 s
`Colo181`	09:32:26.763 → 09:34:44.486	137.7 s	150.1 s

The worker-readiness interval dominates these task durations. That narrows the next investigation to service initialization and its inputs. It does not identify the cause inside the remote process. I did not profile CPU, storage, model loading, or network activity on those hosts. The readiness intervals use Airflow’s log timestamps, so they also include polling and log-delivery delay.

For Colo153, the observed downstream sequence is:

Step on the path	Completed	Task runtime
Stock weight	09:29:55.607	352.6 s
Stock target, `ZX13`	09:30:51.194	52.3 s
Parameters, `HuaTai_ZX13`	09:31:17.144	22.7 s
Transfer to `R7`, `HuaTai_ZX13`	09:31:22.200	1.7 s
Transfer to `Colo153`	09:32:04.378	39.0 s
Last prerequisite account check	09:32:24.517	16.6 s
Shared service ready	09:35:15.291	167.3 s

The transfer and shared-service tasks have multiple upstream accounts. The table follows the latest prerequisite completion at each step, so the account identifier can change between rows. The corresponding stock dependencies are unchanged by the Redis edit in this session.

Stock algo launch and shared-service startup are sibling tasks after the parameter checks. A successful algo-launch task therefore does not establish that the shared worker is ready. Today all 68 stock algo-launch tasks finish by 09:32:47.583, but the last shared-service task finishes at 09:35:15.291. Actual trading readiness needs both, plus the application’s own health signals.

Redis is waiting, rather than running slowly

The nine futures hedge checks finish by 09:10:40.896. The first stock target succeeds at 09:30:02.918, and the last at 09:31:31.982. Redis starts at 09:31:34.769 and runs for 4.8 seconds.

The morning graph waits for every stock target before Redis can run. The revised graph now expresses two separate conditions:

1
2
3


stock targets ── ONE_SUCCESS ── stock_target_pos.first_ready ──┐
                                                            ├── ALL_SUCCESS ── redis_tpc
hedge checks ─── ALL_DONE ───── futures_hedge.checks_done ──────┘

This preserves the existing hedge-check policy: every hedge check must finish, even if one fails. Each futures trading task separately requires its own hedge check and Redis to succeed. Airflow’s trigger-rule documentation explains why ONE_SUCCESS needs its own gate when combined with another condition.

Using today’s timings, the first stock success would satisfy the remaining data condition at 09:30:02.918, since the hedge checks were already finished. Waiting for the last stock target adds 89.1 seconds of dependency delay.

That is an estimate of the dependency wait the new graph can remove. It is not a measured 89.1-second improvement in Redis completion, and it would not make Redis start before 09:30. The two new gate tasks have no execution timestamps in today’s run. Their scheduling overhead and the next run’s completion time still need measurement.

The observed executor queue is a smaller part of today’s delay

Runtime and executor queue delay by task family

Among 308 setup or gate instances with surviving queue timestamps, median queued-to-running delay is 2.7 seconds, p90 is 3.2 seconds, and the maximum is 3.5 seconds. Those values are much smaller than the 353-second stock-weight run or the 150–167-second shared-service startups.

Family	Today’s tasks	Median runtime	Maximum runtime	Finish after 09:30
Stock targets	26	34.3 s	93.3 s	26
Parameter generation	34	22.0 s	22.8 s	34
Host transfer	22	6.4 s	46.2 s	22
Parameter checks	34	10.9 s	22.5 s	34
Shared service startup	22	46.5 s	167.3 s	22
Stock algo launch	68	12.4 s	31.0 s	68
Futures algo launch	9	5.3 s	5.7 s	9
Hedge checks	9	6.7 s	7.0 s	0

Queue time only starts when queued_dttm is set. A task may have waited for a dependency, a scheduler decision, or capacity before that timestamp. These measurements do not rule out every scheduling delay or historical capacity problem.

The current global parallelism and per-DAG active-task limit are both 128. Today’s numbers do not make another limit increase the first optimization. The Nov 11 outage was different: configuration drift and restarts disrupted the session. Its 127 recorded failed morning attempts are shown separately from the duration baseline.

Session-long tasks need a start-time measure

The live stock and futures target-position services intentionally stay alive until about 15:00. Their roughly 330-minute durations are expected behavior, not the slow setup work identified above.

Cumulative starts of live stock and futures session tasks

On Nov 12:

The 34 live stock-session tasks start between 09:30:05.374 and 09:31:34.781.
The nine live futures-session tasks start between 09:30:10.114 and 09:31:35.351.

For these services, I would track launch time, application readiness, heartbeat coverage, and unexpected exits. A short runtime could actually indicate a failure.

What I would optimize next

1. Explain the stock-weight delay in the backend

This task delays every stock target and both downstream setup branches. I would start by comparing backend changes and input sizes around Oct 23–26, when the recorded runtime rises from about 109 seconds to about 341 seconds.

Add or inspect timestamps for request acceptance, backend work start, work completion, and response delivery, using the same request identifier. Those four timestamps distinguish queue delay, computation, and a lost completion response. Profile the component that occupies the time, rather than increasing Airflow parallelism.

The next change depends on that evidence. Reuse stable intermediate results if repeated computation dominates. Reduce storage or network work if reads dominate. If a completion response is lost, fix response delivery and retry handling.

Shortening the client timeout is not a demonstrated speedup. It may send more duplicate requests while the backend is still working. Before changing retry behavior, verify that the same logical request cannot execute its effects twice.

2. Overlap shared-service preparation with account setup where its inputs allow it

The slowest shared task spends 158.7 seconds between readiness polling and its first observed worker message. I would profile startup on Colo153 and Colo152, separating model reads, initialization, external waits, and thread startup.

Then inspect which inputs actually require completed account parameters. If some startup work uses only stable model files, prepare that work earlier and keep a final validation gate before trading uses it. Keep the existing freshness checks and worker-readiness check. A process merely existing is not sufficient readiness.

The potential overlap is bounded by the work that is truly independent. The observed 158.7 seconds is time to investigate, not a promise that all of it can disappear.

This matters for the schedule budget. Today, 319.7 seconds pass between stock-weight completion and the last shared-service success. Keeping that downstream delay unchanged would require stock weight ready around 09:24:40 to finish setup by 09:30. A task that starts after the 09:24 gate and then takes 352.6 seconds leaves too little time. Meeting the deadline needs faster work, safe overlap, or a justified change in input availability, not just a different Redis trigger.

3. Validate the Redis gate on the next scheduled run

The graph change is already in the DAG. Check that Redis starts only after both the first successful stock target and all finished hedge checks, then verify that each futures trading task still waits for its own successful hedge check.

Record gate completion, Redis queue time, Redis start, and futures startup completion. Compare dependency wait separately from runtime. Today’s 89.1-second gap is the estimate to test, not a result to publish as a measured speedup.

4. Investigate the four slow stock targets and the longest transfers

For DC5, ZX14, ZX8, and ZZ1, correlate the 60-second client retries with backend request records. The nearly simultaneous task starts and staggered completions justify checking backend serialization or contention, but do not prove either.

The path to Colo153 also includes 22.7 seconds of parameter generation and 39.0 seconds of host transfer. Measure bytes transferred, remote command time, and repeated work for that path. Reuse unchanged files or intermediate results where freshness and validation remain intact. Do not remove account checks just to make the graph shorter.

The inspected image-pull log lines report already-current images within fractions of a second. Those selected pulls are a lower priority than the application waits. There is no evidence here for a large gain from changing image-pull policy.

5. Measure joint readiness and retain attempt history

Record a readiness timestamp per account and host that requires both the algo launch and its shared service. Alert on missed readiness at 09:30, and distinguish an unresolved task from a successful task that ran late. Add application-level first-use and heartbeat signals so task state is not the only evidence of a working service.

Retain attempt history and task logs outside the container’s writable layer. Today’s reconstruction recovered earlier attempts, but 93 first morning outcomes remain unresolved across the historical sample. Better records would make the next comparison more reliable.

Keep startup capacity visible while the 43 session tasks occupy slots. Airflow pools can cap related tasks, but a pool alone does not reserve global executor capacity. Change limits only after measuring runnable work waiting for capacity. With LocalExecutor, restarting the scheduler can also orphan running tasks, so a restart is not a harmless way to test an optimization during the session.

For each proposed change, collect several comparable scheduled runs with the same input-freshness requirements. Report per-stage readiness and failures together. The next observation I want is whether stock weight still consumes almost the entire six-minute interval from 09:24 to 09:30, and whether shared workers become ready earlier without weakening validation.

Reproduce the figures

The bundle includes analyze.py, attempts.csv, and metadata.json. Download them into the same directory and run:

1

python analyze.py

The script needs pandas, numpy, and matplotlib. It reads the recorded snapshot locally and does not connect to production. It regenerates all six figures and the following tables:

Daily readiness, in seconds relative to 09:30.
Family summary, with historical sample counts and queue coverage.
Every observed first morning setup task on Nov 12.
Headline numbers.

The supporting evidence includes request phases, readiness-poll intervals, selected original log lines, and current run-level timestamps. The last file deliberately retains the misleading afternoon run timestamps so the rerun limitation is visible.

The production inspection was read-only. I did not profile the backend or remote trading hosts, rerun production tasks to benchmark them, or measure a before-and-after improvement from the new Redis gates. Those are the remaining checks behind the optimization suggestions.

Complete reproduction code and guide

Download the complete reproduction bundle, which contains the scripts, README.md, the original timing snapshot, and log evidence. The runner reproduces the saved results without credentials. Its optional --refresh mode reads Airflow over SSH; a password, when needed, comes only from AIRFLOW_SSH_PASSWORD in the environment.

1
2
3


unzip reproduce.zip -d hyena17-report
cd hyena17-report
bash run.sh

The analysis script, read-only exporter, and Bash runner are also available separately. Install the recorded dependencies as described in README.md before running. All source files follow below.

analyze.py

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330


"""Regenerate this post's figures and tables from its snapshot with original Airflow identifiers.

Run: python analyze.py
Requires pandas, numpy, and matplotlib. Does not connect to Airflow.
"""
import json
from pathlib import Path

import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
from matplotlib.colors import TwoSlopeNorm
import numpy as np
import pandas as pd

ROOT = Path(__file__).resolve().parent
TZ = 'Asia/Shanghai'
INCIDENT = '2026-11-11'
META = json.loads((ROOT / 'metadata.json').read_text())
LATEST = max(META['scheduled_dates'])
BLUE, RED, GOLD, GREEN = '#2563a6', '#c2413b', '#d49b32', '#138575'
plt.rcParams.update({'font.family': 'DejaVu Sans', 'font.size': 10, 'axes.spines.top': False,
                     'axes.spines.right': False, 'axes.titleweight': 'bold', 'savefig.facecolor': 'white'})


def family(task):
    for prefix, label in [
        ('zsun_live_trade_stock_target_pos_', 'Live stock session'),
        ('zsun_live_trade_futures_target_pos_', 'Live futures session'),
        ('zsun_pre_trade_stock_target_pos_', 'Stock targets'),
        ('zsun_pre_trade_futures_target_pos_', 'Futures targets'),
        ('zfu_param_to_R7.', 'Relay transfer'),
        ('zfu_param_to_', 'Host transfer'),
        ('zfu_param_', 'Parameter generation'),
        ('zfu_check_', 'Parameter checks'),
        ('h17.trading.shared_', 'Shared service startup'),
        ('h17.trading.raven_snap_', 'Snapshot service startup'),
        ('h17.trading.', 'Stock algo launch'),
        ('futures_hedge.check.', 'Hedge checks'),
        ('futures_hedge.trading.', 'Futures algo launch'),
        ('wait_until_', 'Intentional wait'),
    ]:
        if task.startswith(prefix):
            return label
    return {'zsun_pre_trade_stock_weight': 'Stock weight', 'redis_tpc': 'Redis startup',
            'futures_hedge.risk': 'Futures risk', 'futures_hedge.param': 'Futures parameters'}.get(task, 'Other setup')


def save(fig, name):
    fig.savefig(ROOT / name, dpi=165, bbox_inches='tight')
    plt.close(fig)


def clock(seconds):
    return (pd.Timestamp('2000-01-01 09:30') + pd.Timedelta(seconds=float(seconds))).strftime('%H:%M:%S')


def short(task):
    for prefix, label in [
        ('zsun_pre_trade_stock_target_pos_', 'Stock target position'),
        ('h17.trading.shared_', 'Shared services'),
        ('h17.trading.', 'Stock algo launch'),
        ('zfu_param_to_R7.', 'zfu_param_to_R7'),
        ('zfu_param_to_', 'zfu_param_to'),
        ('zfu_param_', 'zfu_param'),
        ('zfu_check_', 'zfu_check'),
        ('futures_hedge.trading.', 'Futures launch'),
    ]:
        if task.startswith(prefix):
            name = task[len(prefix):]
            return f'{label}: {name}'
    return task


def running_seconds_after_deadline(started, ended):
    cutoff = started.dt.normalize() + pd.Timedelta(hours=9, minutes=30)
    runtime = (ended - started).dt.total_seconds()
    late = (ended - cutoff).dt.total_seconds().clip(lower=0)
    return np.minimum(runtime, late)


def load_attempts():
    df = pd.read_csv(ROOT / 'attempts.csv')
    for col in ['started_at', 'ended_at']:
        df[col] = pd.to_datetime(df[col], utc=True).dt.tz_convert(TZ)
    df['family'] = df.task_id.map(family)
    cutoff = df.started_at.dt.normalize() + pd.Timedelta(hours=9, minutes=30)
    df['runtime_s'] = (df.ended_at - df.started_at).dt.total_seconds()
    df['finish_offset_s'] = (df.ended_at - cutoff).dt.total_seconds()
    df['start_offset_s'] = (df.started_at - cutoff).dt.total_seconds()
    df['running_after_s'] = running_seconds_after_deadline(df.started_at, df.ended_at)
    assert df.loc[df.ended_at.notna(), 'runtime_s'].ge(0).all(), 'Negative duration'
    assert df.queue_seconds.dropna().ge(0).all(), 'Negative queue delay'
    assert df.running_after_s.dropna().ge(0).all()
    assert (df.running_after_s.dropna() <= df.runtime_s.dropna() + 0.001).all()
    return df


def run():
    df = load_attempts()
    morning = df[df.started_at.dt.strftime('%Y-%m-%d').eq(df.date)
                 & df.started_at.dt.hour.ge(9) & df.started_at.dt.hour.lt(12)].copy()
    first = morning.sort_values('started_at').drop_duplicates(['date', 'task_id'])
    success = first[first.state.eq('success')].copy()
    latest = success[success.date.eq(LATEST)].copy()
    session = latest.family.str.startswith('Live ')
    setup = latest[~session & ~latest.family.eq('Intentional wait')].copy()
    available_dates = sorted(success.loc[success.family.eq('Stock weight'), 'date'].unique())
    baseline = success[success.date.isin(set(available_dates) - {INCIDENT})]

    # Readiness uses the first successful morning attempt, including recovery attempts.
    recovered = morning[morning.state.eq('success')].sort_values('ended_at').drop_duplicates(['date', 'task_id'])
    history = []
    for day in META['scheduled_dates']:
        g = recovered[recovered.date.eq(day)]
        def offset(label, agg='max'):
            v = g.loc[g.family.eq(label), 'finish_offset_s']
            return getattr(v, agg)() if len(v) else np.nan
        history.append({'date': day, 'Stock weight done': offset('Stock weight'),
                        'First stock target': offset('Stock targets', 'min'),
                        'Last stock target': offset('Stock targets'), 'Redis done': offset('Redis startup'),
                        'Last algo launch': offset('Stock algo launch'), 'Last shared ready': offset('Shared service startup')})
    history = pd.DataFrame(history).set_index('date')
    history.to_csv(ROOT / 'daily-readiness.csv', float_format='%.3f')

    # Latest morning timeline, with queues shown only when the original queue timestamp survived.
    stock = latest[latest.family.eq('Stock targets')].sort_values('ended_at')
    slow_shared = latest[latest.family.eq('Shared service startup')].nlargest(4, 'runtime_s')
    algo = latest[latest.family.eq('Stock algo launch')].sort_values('task_id')
    shared = latest[latest.family.eq('Shared service startup')].sort_values('task_id')
    futures = latest[latest.family.eq('Futures algo launch')].sort_values('task_id')
    params = latest[latest.family.eq('Parameter generation')].sort_values('task_id')
    relay = latest[latest.family.eq('Relay transfer')].sort_values('task_id')
    transfers = latest[latest.family.eq('Host transfer')].sort_values('task_id')
    checks = latest[latest.family.eq('Parameter checks')].sort_values('task_id')
    account_colos = algo.task_id.str.removeprefix('h17.trading.').str.split('_', n=1)
    accounts = {parts[1] for parts in account_colos}
    colos = {parts[0] for parts in account_colos}
    assert accounts == set(latest.loc[latest.family.eq('Live stock session'), 'task_id'].str.removeprefix('zsun_live_trade_stock_target_pos_'))
    assert colos == set(shared.task_id.str.removeprefix('h17.trading.shared_'))
    assert set(futures.task_id.str.removeprefix('futures_hedge.trading.')) == set(latest.loc[latest.family.eq('Live futures session'), 'task_id'].str.removeprefix('zsun_live_trade_futures_target_pos_'))
    assert accounts == set(params.task_id.str.removeprefix('zfu_param_'))
    assert accounts == set(relay.task_id.str.removeprefix('zfu_param_to_R7.'))
    assert accounts == set(checks.task_id.str.removeprefix('zfu_check_'))
    assert colos == set(transfers.task_id.str.removeprefix('zfu_param_to_'))
    panels = [
        (f'{LATEST}: stock weight and all {len(stock)} fund targets', pd.concat([latest[latest.family.eq('Stock weight')], stock])),
        (f'zfu_param generation: all {len(params)} accounts', params),
        (f'zfu_param_to_R7 transfers: all {len(relay)} accounts', relay),
        (f'zfu_param_to_Colo transfers: all {len(transfers)} colos', transfers),
        (f'zfu_check validation: all {len(checks)} accounts', checks),
        (f'Stock algo launches: {len(accounts)} accounts, {len(colos)} colos, {len(algo)} account/colo tasks', algo),
        (f'Shared services: all {len(shared)} colos', shared),
        (f'Redis, all {len(futures)} futures account launches, and risk', pd.concat([latest[latest.family.eq('Redis startup')], futures, latest[latest.family.eq('Futures risk')]])),
    ]
    timeline_families = ['Stock weight', 'Stock targets', 'Parameter generation', 'Relay transfer',
                         'Host transfer', 'Parameter checks', 'Stock algo launch',
                         'Shared service startup', 'Redis startup', 'Futures algo launch', 'Futures risk']
    expected_tasks = set(latest.loc[latest.family.isin(timeline_families), 'task_id'])
    drawn_tasks = [task for _, rows in panels for task in rows.task_id]
    assert len(drawn_tasks) == len(set(drawn_tasks)) and set(drawn_tasks) == expected_tasks
    fig, axes = plt.subplots(len(panels), 1, figsize=(16, sum(len(rows) for _, rows in panels)*.24+5),
                             gridspec_kw={'height_ratios': [len(rows)+3 for _, rows in panels]})
    for ax, (title, rows) in zip(axes, panels):
        ax.axvspan(0, 6.1, color=RED, alpha=.045)
        for y, row in enumerate(rows.itertuples()):
            start = row.start_offset_s / 60
            width = row.runtime_s / 60
            if pd.notna(row.queue_seconds):
                ax.barh(y, row.queue_seconds / 60, left=start-row.queue_seconds/60, height=.55, color=GOLD)
            ax.barh(y, width, left=start, height=.55, color=BLUE)
            ax.text(start+width+.04, y, f'{row.runtime_s:.1f}s', va='center', fontsize=8.5)
        ax.set_yticks(range(len(rows)), [short(s) for s in rows.task_id], fontsize=9)
        ax.invert_yaxis()
        ax.axvline(0, color=RED, lw=1.4)
        ax.set_xlim(-6.2, 6.05)
        ax.set_xticks(np.arange(-6, 7, 1), [clock(x*60)[:5] for x in range(-6, 7)])
        ax.set_title(title, loc='left', pad=14)
        ax.grid(axis='x', alpha=.18)
    axes[-1].set_xlabel('Local time, UTC+8. Blue = task running; amber = queued to running, where available.')
    fig.tight_layout()
    save(fig, 'morning-timeline.png')

    # Historical completion times, not summed parallel task durations.
    values = history.to_numpy(dtype=float)/60
    fig, ax = plt.subplots(figsize=(12, 10.2))
    cmap = plt.get_cmap('RdYlBu_r').copy()
    cmap.set_bad('#e2e5e9')
    image = ax.imshow(np.ma.masked_invalid(values), aspect='auto', cmap=cmap,
                      norm=TwoSlopeNorm(vmin=-10, vcenter=0, vmax=10))
    for y in range(len(history)):
        for x in range(len(history.columns)):
            value = history.iloc[y, x]
            ax.text(x, y, clock(value) if np.isfinite(value) else 'no record', ha='center', va='center', fontsize=8.5,
                    color='white' if np.isfinite(value) and abs(value)>480 else '#18212b')
    ax.set_xticks(range(len(history.columns)), [s.replace(' ', '\n', 1) for s in history.columns])
    ax.set_yticks(range(len(history)), [d + ('  * incident' if d == INCIDENT else '') for d in history.index])
    ax.set_title('Twenty scheduled dates: when did each stage finish?', loc='left', pad=18)
    ax.text(0, -.11, 'UTC+8. Gray = missing successful morning record, not an on-time completion.\n'
            'Color clips at +/-10 minutes; printed timestamps retain the full delay. Incident day is separate from baseline.',
            transform=ax.transAxes, fontsize=9)
    fig.colorbar(image, ax=ax, shrink=.8, label='Minutes relative to 09:30')
    fig.tight_layout()
    save(fig, 'readiness-history.png')

    # Rank own runtime after the deadline. This is not end-to-end lateness or CPU time.
    ranking = setup[setup.running_after_s.gt(0)].nlargest(14, 'running_after_s').sort_values('running_after_s')
    historical = baseline.groupby('task_id').running_after_s.quantile(.9)
    fig, ax = plt.subplots(figsize=(12, 7.8))
    ax.barh(range(len(ranking)), ranking.running_after_s, color=BLUE, height=.6, label=f'{LATEST} first attempt')
    p90 = [historical.get(t, np.nan) for t in ranking.task_id]
    ax.scatter(p90, np.arange(len(ranking))-.23, color=RED, marker='D', s=27, label='Historical p90, successful first attempts')
    for y, row in enumerate(ranking.itertuples()):
        ax.text(row.running_after_s+3, y, f'{row.running_after_s:.1f}s; done {row.ended_at:%H:%M:%S}', va='center', fontsize=8)
    ax.set_yticks(range(len(ranking)), [short(t) for t in ranking.task_id], fontsize=9)
    ax.set_xlim(0, max(ranking.running_after_s.max(), np.nanmax(p90))*1.34)
    ax.set_xlabel('Seconds this task was running after 09:30, excluding queue and upstream waits')
    ax.set_title('The longest setup tasks after the open are shared-service startups', loc='left', pad=18)
    ax.legend(loc='lower right', fontsize=8)
    ax.grid(axis='x', alpha=.18)
    fig.tight_layout()
    save(fig, 'late-task-ranking.png')

    # Dependency readiness and queue delay are different intervals.
    families = ['Stock weight', 'Stock targets', 'Parameter generation', 'Host transfer', 'Parameter checks',
                'Shared service startup', 'Stock algo launch', 'Redis startup', 'Futures algo launch', 'Hedge checks']
    family_rows=[]
    for label in families:
        g=setup[setup.family.eq(label)]
        b=baseline[baseline.family.eq(label)]
        family_rows.append({'family':label, 'latest_n':len(g), 'latest_runtime_p50_s':g.runtime_s.median(),
                            'latest_runtime_max_s':g.runtime_s.max(), 'latest_late_n':int(g.finish_offset_s.gt(0).sum()),
                            'latest_queue_n':int(g.queue_seconds.notna().sum()), 'latest_queue_p50_s':g.queue_seconds.median(),
                            'latest_queue_max_s':g.queue_seconds.max(), 'historical_n':len(b),
                            'historical_runtime_p50_s':b.runtime_s.median(), 'historical_runtime_p90_s':b.runtime_s.quantile(.9)})
    family_table=pd.DataFrame(family_rows)
    family_table.to_csv(ROOT/'family-summary.csv',index=False,float_format='%.3f')
    fig, axes = plt.subplots(1, 2, figsize=(12, 6.6))
    for ax, column, title, color in [(axes[0],'latest_runtime_p50_s','Median task runtime',BLUE),
                                    (axes[1],'latest_queue_p50_s','Median queued-to-running delay',GOLD)]:
        ax.barh(family_table.family, family_table[column], color=color, height=.6)
        ax.invert_yaxis()
        ax.set_title(title,loc='left')
        ax.set_xlabel('Seconds')
        ax.grid(axis='x',alpha=.18)
        for y, value in enumerate(family_table[column]):
            if pd.notna(value):
                ax.text(value+family_table[column].max()*.015,y,f'{value:.1f}',va='center',fontsize=9)
        ax.margins(x=.15)
    axes[1].tick_params(labelleft=False)
    fig.suptitle(f'{LATEST}: observed executor queue delays are much smaller than task runtime',x=.02,ha='left',fontweight='bold')
    fig.text(.02,.005,'Queue timestamps survive only for matching current task instances. Time before queueing is not measured here.',fontsize=9)
    fig.tight_layout(rect=[0,.03,1,.95])
    save(fig,'runtime-and-queue.png')

    # This is an observational change in recorded timings, not a controlled benchmark.
    weights = recovered[recovered.family.eq('Stock weight')].sort_values('date')
    fig, ax = plt.subplots(figsize=(11.5, 4.8))
    xpos = np.arange(len(weights))
    colors = [RED if d == INCIDENT else BLUE for d in weights.date]
    ax.bar(xpos, weights.runtime_s, color=colors, width=.65)
    ax.axhline(300, color=GOLD, ls='--', lw=1.4, label='300s request timeout seen in Nov 11-12 logs')
    ax.set_xticks(xpos, [d[5:] for d in weights.date], rotation=45, ha='right')
    ax.set_ylabel('Task runtime, seconds')
    ax.set_xlabel('Scheduled date, 2026. Earlier task logs are unavailable in the rebuilt container.')
    ax.set_title('Stock-weight time rises from about 109s to about 341s after Oct 23', loc='left', pad=15)
    ax.legend(loc='upper left', fontsize=8)
    ax.grid(axis='y', alpha=.18)
    fig.tight_layout()
    save(fig, 'stock-weight-history.png')

    # Long-lived tasks get a launch-time measure instead of a five-hour "slow" flag.
    live = first[first.date.eq(LATEST) & first.family.str.startswith('Live ')].copy()
    fig, axes=plt.subplots(2,1,figsize=(11,5.8),sharex=True)
    for ax,label in zip(axes,['Live stock session','Live futures session']):
        g=live[live.family.eq(label)].sort_values('start_offset_s')
        x=np.sort(np.r_[0,g.start_offset_s.to_numpy()/60])
        ax.step(x,np.arange(len(x)),where='post',color=GREEN,lw=2)
        ax.scatter(g.start_offset_s/60,np.arange(1,len(g)+1),s=14,color=GREEN)
        ax.set_ylabel('Tasks started')
        ax.set_title(f'{label}: {len(g)} starts, {clock(g.start_offset_s.min())} to {clock(g.start_offset_s.max())}',loc='left')
        ax.grid(alpha=.18)
    axes[-1].set_xlabel('Minutes after 09:30. A task start does not prove the first order or downstream service readiness.')
    axes[-1].set_xlim(0,max(live.start_offset_s)/60*1.08)
    fig.tight_layout()
    save(fig,'live-session-starts.png')

    setup.sort_values('finish_offset_s',ascending=False)[['task_id','family','started_at','ended_at','runtime_s','running_after_s','queue_seconds','source']].to_csv(ROOT/'latest-morning-tasks.csv',index=False,float_format='%.3f')
    stock_weight=latest[latest.family.eq('Stock weight')].iloc[0]
    redis=latest[latest.family.eq('Redis startup')].iloc[0]
    checks=latest[latest.family.eq('Hedge checks')]
    first_stock=stock.ended_at.min()
    last_stock=stock.ended_at.max()
    eligible=max(first_stock,checks.ended_at.max())
    summary={'date_range':[min(META['scheduled_dates']),LATEST], 'scheduled_dates':len(META['scheduled_dates']),
             'baseline_dates':len(set(available_dates)-{INCIDENT}), 'first_attempt_states':first.state.value_counts().to_dict(),
             'latest_first_attempt_states':first[first.date.eq(LATEST)].state.value_counts().to_dict(),
             'stock_weight_runtime_s':stock_weight.runtime_s,'stock_weight_finish':str(stock_weight.ended_at),
             'stock_weight_headroom_s':-stock_weight.finish_offset_s,'stock_targets':len(stock),
             'first_stock_done':str(first_stock),'last_stock_done':str(last_stock),
             'stock_target_barrier_s':(last_stock-first_stock).total_seconds(), 'redis_started':str(redis.started_at),
             'redis_runtime_s':redis.runtime_s, 'last_hedge_check_done':str(checks.ended_at.max()),
             'new_redis_eligibility_lower_bound':str(eligible), 'latest_setup_tasks':len(setup),
             'latest_late_setup_tasks':int(setup.finish_offset_s.gt(0).sum()),
             'latest_setup_queue_p50_s':setup.queue_seconds.median(), 'latest_setup_queue_p90_s':setup.queue_seconds.quantile(.9),
             'latest_setup_queue_max_s':setup.queue_seconds.max(),'latest_queue_samples':int(setup.queue_seconds.notna().sum()),
             'latest_running_session_tasks':len(live),
             'live_session_launches':{label:{'count':len(g),'first':str(g.started_at.min()),'last':str(g.started_at.max())} for label,g in live.groupby('family')},
             'latest_slowest_shared':slow_shared[['task_id','runtime_s','ended_at']].astype({'ended_at':str}).to_dict('records'),
             'missing_stock_weight_dates':sorted(set(META['scheduled_dates'])-set(available_dates)),
             'incident_morning_failures':int(morning[morning.date.eq(INCIDENT)].state.eq('failed').sum())}
    (ROOT/'summary.json').write_text(json.dumps(summary,indent=2,default=str)+'\n')
    print(json.dumps(summary,indent=2,default=str))
    print('\nFAMILY SUMMARY\n'+family_table.round(2).to_string(index=False))


if __name__=='__main__':
    assert short('zsun_pre_trade_stock_target_pos_ZX7') == 'Stock target position: ZX7'
    assert short('h17.trading.shared_Colo153') == 'Shared services: Colo153'
    assert family('h17.trading.Colo153_Broker_Account') == 'Stock algo launch'
    assert family('zfu_param_to_Colo153') == 'Host transfer'
    assert family('zfu_param_to_R7.Broker_Account') == 'Relay transfer'
    starts = pd.Series(pd.to_datetime(['2026-11-12T01:28Z', '2026-11-12T01:29Z',
                                      '2026-11-12T01:31Z', '2026-11-12T01:31Z'])).dt.tz_convert(TZ)
    ends = pd.Series(pd.to_datetime(['2026-11-12T01:29Z', '2026-11-12T01:31Z',
                                    '2026-11-12T01:33Z', None])).dt.tz_convert(TZ)
    overlap = running_seconds_after_deadline(starts, ends)
    assert overlap.iloc[:3].tolist() == [0, 60, 120]
    assert pd.isna(overlap.iloc[3]), 'An unfinished task must not become a zero-duration success'
    run()

export_airflow.py

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163


"""Read-only export of the latest 20 completed trading.hyena17 scheduled runs.

Run inside the existing Airflow container: python export_airflow.py
Save that output locally, then run: python export_airflow.py --write-snapshot OUTPUT
Only task timing and non-secret configuration fields are exported.
"""
import csv
from datetime import datetime, timezone
import io
import json
import os
from pathlib import Path
import sys
import tempfile

DAG_ID = 'trading.hyena17'
PREFIX = 'AIRFLOW_TIMING_JSON='
FIELDS = ['date', 'task_id', 'attempt', 'started_at', 'ended_at', 'state', 'queue_seconds', 'source']


def utc(value):
    if value is None:
        return None
    if isinstance(value, str):
        value = datetime.fromisoformat(value)
    return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value.astimezone(timezone.utc)


def reconstruct(events, tasks, failures):
    rows, pending, unmatched = [], {}, 0
    for event in events:
        key = event['execution_date'], event['task_id']
        if event['event'] == 'running':
            if key in pending:
                rows.append(pending.pop(key))
            pending[key] = {'date': str(event['execution_date'])[:10], 'task_id': event['task_id'],
                            'started_at': utc(event['dttm']), 'ended_at': None, 'state': 'unknown',
                            'queue_seconds': None, 'source': 'audit_log_unfinished'}
        elif key in pending:
            row = pending.pop(key)
            row.update(ended_at=utc(event['dttm']), state=event['event'], source='audit_log')
            rows.append(row)
        else:
            unmatched += 1
    rows.extend(pending.values())
    lookup = {}
    for row in rows:
        lookup.setdefault((row['date'], row['task_id']), []).append(row)
    for records, source in [(tasks, 'task_instance'), (failures, 'task_fail')]:
        for task in records:
            start, end = utc(task['start_date']), utc(task['end_date'])
            if start is None:
                continue
            day = task['run_id'][len('scheduled__'):len('scheduled__') + 10]
            state = task.get('state', 'failed')
            candidates = lookup.get((day, task['task_id']), [])
            match = next((row for row in candidates if abs((row['started_at'] - start).total_seconds()) < 1
                          and ((row['state'] == 'unknown' and row['ended_at'] is None)
                               or (row['state'] == state and row['ended_at'] and end
                                   and abs((row['ended_at'] - end).total_seconds()) < 1))), None)
            if match is None:
                if end is None or state not in ['success', 'failed'] or task.get('operator') != 'EmptyOperator':
                    continue
                match = {'date': day, 'task_id': task['task_id'], 'queue_seconds': None}
                rows.append(match)
                lookup.setdefault((day, task['task_id']), []).append(match)
            match.update(started_at=start, ended_at=end, state=state, source=source)
            queued = utc(task.get('queued_dttm'))
            if queued and queued <= start:
                match['queue_seconds'] = (start - queued).total_seconds()
    rows.sort(key=lambda row: (row['date'], row['task_id'], row['started_at']))
    counts = {}
    for row in rows:
        key = row['date'], row['task_id']
        counts[key] = counts.get(key, 0) + 1
        row['attempt'] = counts[key]
        row['started_at'] = row['started_at'].isoformat()
        row['ended_at'] = row['ended_at'].isoformat() if row['ended_at'] else ''
    return rows, unmatched


def capture():
    import airflow
    from airflow import settings
    from airflow.configuration import conf
    from sqlalchemy import bindparam, text

    engine = settings.engine
    with engine.connect() as conn:
        if engine.dialect.name == 'mysql':
            conn.execute(text('SET SESSION TRANSACTION READ ONLY'))
        def read(query, values):
            return [dict(row._mapping) for row in conn.execute(query, values)]
        runs = read(text("SELECT run_id,execution_date,start_date,end_date,state,dag_hash FROM dag_run "
                         "WHERE dag_id=:dag AND run_type='scheduled' AND state IN ('success','failed') "
                         "ORDER BY execution_date DESC LIMIT 20"), {'dag': DAG_ID})
        if not runs:
            raise ValueError('No completed scheduled runs found')
        params = {'dag': DAG_ID, 'runs': [row['run_id'] for row in runs]}
        def task_rows(table, columns):
            query = text('SELECT ' + columns + ' FROM ' + table
                         + ' WHERE dag_id=:dag AND run_id IN :runs').bindparams(bindparam('runs', expanding=True))
            return read(query, params)
        tasks = task_rows('task_instance', 'run_id,task_id,state,start_date,end_date,queued_dttm,operator')
        failures = task_rows('task_fail', 'run_id,task_id,start_date,end_date')
        query = text("SELECT dttm,task_id,event,execution_date FROM log WHERE dag_id=:dag "
                     "AND execution_date IN :dates AND event IN ('running','success','failed') "
                     "ORDER BY execution_date,task_id,dttm,id").bindparams(bindparam('dates', expanding=True))
        events = read(query, {'dag': DAG_ID, 'dates': [row['execution_date'] for row in runs]})
    attempts, unmatched = reconstruct(events, tasks, failures)
    metadata = {'collected_at_utc': datetime.now(timezone.utc).isoformat(), 'dag_id': DAG_ID,
                'airflow_version': airflow.__version__, 'timezone': conf.get('core', 'default_timezone'),
                'executor': conf.get('core', 'executor'),
                'config': {key: conf.get(section, key) for section, key in [
                    ('core', 'parallelism'), ('core', 'max_active_tasks_per_dag'),
                    ('core', 'max_active_runs_per_dag'), ('scheduler', 'scheduler_heartbeat_sec')]},
                'scheduled_dates': sorted(str(row['execution_date'])[:10] for row in runs),
                'dag_hashes': len({row['dag_hash'] for row in runs}), 'current_task_rows': len(tasks),
                'failure_rows': len(failures), 'audit_event_count': len(events),
                'ignored_terminals_without_start': unmatched}
    run_rows = [{'date': str(row['execution_date'])[:10], 'state': row['state'],
                 'metadata_start_utc': utc(row['start_date']).isoformat() if row['start_date'] else '',
                 'metadata_end_utc': utc(row['end_date']).isoformat() if row['end_date'] else ''}
                for row in reversed(runs)]
    return {'metadata': metadata, 'attempts': attempts, 'runs': run_rows}


def csv_text(rows, fields):
    buffer = io.StringIO(newline='')
    writer = csv.DictWriter(buffer, fieldnames=fields)
    writer.writeheader()
    writer.writerows(rows)
    return buffer.getvalue()


def write_snapshot(path):
    lines = [line[len(PREFIX):] for line in Path(path).read_text().splitlines() if line.startswith(PREFIX)]
    if len(lines) != 1:
        raise ValueError('Expected exactly one complete timing snapshot in the SSH output')
    snapshot = json.loads(lines[0])
    if snapshot['metadata']['dag_id'] != DAG_ID or not snapshot['attempts']:
        raise ValueError('Wrong DAG or empty task snapshot')
    contents = {
        'attempts.csv': csv_text(snapshot['attempts'], FIELDS),
        'metadata.json': json.dumps(snapshot['metadata'], indent=2) + '\n',
        'runs.csv': csv_text(snapshot['runs'], ['date', 'state', 'metadata_start_utc', 'metadata_end_utc']),
    }
    root = Path(__file__).resolve().parent
    for name, content in contents.items():
        with tempfile.NamedTemporaryFile(mode='w', dir=root, prefix='.' + name, delete=False) as handle:
            handle.write(content)
            temporary = handle.name
        os.replace(temporary, root / name)
    print('Saved timing snapshot for ' + snapshot['metadata']['scheduled_dates'][-1])


if __name__ == '__main__':
    if len(sys.argv) == 3 and sys.argv[1] == '--write-snapshot':
        write_snapshot(sys.argv[2])
    elif len(sys.argv) == 1:
        print(PREFIX + json.dumps(capture()))
    else:
        raise SystemExit('Usage: export_airflow.py [--write-snapshot SSH_OUTPUT]')

run.sh

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28


#!/usr/bin/env bash
set +x
set -euo pipefail
cd -- "$(dirname -- "${BASH_SOURCE[0]}")"
python_bin="${PYTHON:-python3}"
if [[ $# -gt 1 || ( $# -eq 1 && "$1" != --refresh ) ]]; then
    echo 'Usage: ./run.sh [--refresh]' >&2
    exit 2
fi
if [[ "${1:-}" == --refresh ]]; then
    ssh_user="${AIRFLOW_SSH_USER:-infra}"
    ssh_host="${AIRFLOW_SSH_HOST:-infra.t1}"
    snapshot_file="$(mktemp)"
    trap 'rm -f -- "$snapshot_file"' EXIT
    ssh_command=(ssh -o ConnectTimeout=15)
    if [[ -n "${AIRFLOW_SSH_PASSWORD:-}" ]]; then
        command -v sshpass >/dev/null || { echo 'Password authentication needs sshpass.' >&2; exit 1; }
        ssh_command=(sshpass -e ssh -o ConnectTimeout=15 -o NumberOfPasswordPrompts=1)
    else
        ssh_command+=(-o BatchMode=yes)
    fi
    SSHPASS="${AIRFLOW_SSH_PASSWORD:-}" "${ssh_command[@]}" "$ssh_user@$ssh_host" \
        'docker exec -i airflow /usr/local/python3/bin/python3 -' \
        < export_airflow.py > "$snapshot_file"
    unset AIRFLOW_SSH_PASSWORD
    "$python_bin" export_airflow.py --write-snapshot "$snapshot_file"
fi
"$python_bin" analyze.py

README.md

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58


# Reproduce the trading.hyena17 timing report

`./run.sh` regenerates all six plots and summary tables from the included snapshot. It does not contact Airflow or need credentials. The first chart contains all 252 selected startup tasks, including parameter generation, transfers, and checks.

## Install and run

Use Python 3.11. The original environment used Python 3.11.4 with these package versions:

```bash
python3.11 -m venv .venv
.venv/bin/python -m pip install pandas==1.5.2 numpy==1.26.4 matplotlib==3.10.6
PYTHON=.venv/bin/python bash run.sh
```

Run from the directory containing `run.sh`, `analyze.py`, `attempts.csv`, and `metadata.json`. The script also works when invoked by its absolute path. If the dependencies are already installed, `bash run.sh` is sufficient.

The historical snapshot covers 20 scheduled dates from 2026-10-16 through 2026-11-12. Times are converted from UTC to Asia/Shanghai, UTC+8. No tasks are rerun. The included task and account names are the original Airflow identifiers.

## Files and outputs

- `analyze.py`: the complete plotting and summary script, including deadline-overlap and coverage checks.
- `export_airflow.py`: read-only metadata export and reconstruction of earlier task attempts from audit events.
- `attempts.csv`, `metadata.json`, and `runs.csv`: the recorded timing snapshot.
- Six PNG figures, including `morning-timeline.png`, the full 252-row startup timeline.
- `daily-readiness.csv`, `family-summary.csv`, `latest-morning-tasks.csv`, and `summary.json`: generated tables and headline values.
- `request-phases.csv`, `readiness-phases.csv`, and `log-evidence.txt`: original Nov 12 log evidence. These files are separate from the plotting inputs.

## Refresh from Airflow

The default connection is the SSH user `infra` on `infra.t1`. The remote user needs permission to read the existing `airflow` container through Docker. The exporter uses that container's configured metadata connection; it never prints the connection URL or its credentials.

Copy the bundle to a separate directory before refreshing if you want to retain the original results. After a completed trading session, use SSH key authentication:

```bash
export AIRFLOW_SSH_USER=infra
export AIRFLOW_SSH_HOST=infra.t1
PYTHON=.venv/bin/python bash run.sh --refresh
```

If password authentication is required, set `AIRFLOW_SSH_PASSWORD` through your environment or secret manager. In an interactive Bash shell, enter it without showing or storing it in shell history:

```bash
read -r -s -p 'SSH password: ' AIRFLOW_SSH_PASSWORD
printf '\n'
export AIRFLOW_SSH_PASSWORD
PYTHON=.venv/bin/python bash run.sh --refresh
unset AIRFLOW_SSH_PASSWORD
```

This mode needs `sshpass`. It passes the password through the environment rather than a command-line argument. The runner disables shell tracing before handling credentials. No password values or database credentials belong in these files.

Refresh selects the latest 20 completed scheduled runs for `trading.hyena17`, reads task instances, failure records, and audit events, and writes `attempts.csv`, `metadata.json`, and `runs.csv` locally. The MySQL session is read-only. It does not clear tasks, trigger DAGs, restart services, or change the remote database.

The plotting script then regenerates the figures and summary tables. Fresh results can differ from the Nov 12 report. The article text and the three original log-evidence files are not refreshed automatically. An incomplete or changed task layout can fail the account/colo coverage checks instead of silently producing a partial chart.

## Interpretation limits

Only successful first morning attempts enter duration percentiles. Failed and unresolved attempts remain in the snapshot. Missing data is not treated as a zero-duration success. Queue measurements require a surviving matching task-instance record. Long-lived trading sessions are measured by their start times rather than treated as slow setup tasks.