wasde-ingestor

<essential_principles> WASDE Ingestor Core Principles

1. Commodity Scope

WASDE covers three major categories of commodities, each with different table structures and units:

Category	Commodity	US Units	World Units
Grains	Wheat, Corn, Rice, Barley, Sorghum, Oats	million bushels / cwt	million metric tons
Oilseeds	Soybeans, Soybean Oil, Soybean Meal	million bushels / pounds	million metric tons
Cotton	Upland, Pima	million 480-lb bales	million 480-lb bales
Livestock	Beef, Pork, Poultry, Eggs, Dairy	million lbs / dozen / lbs	- (US only)
Sugar	Sugar	1,000 short tons	- (US + Mexico only)

2. Table Structure Consistency

All supply and demand tables follow the same balance formula:

Ending Stocks = Beginning Stocks + Production + Imports - Total Use - Exports

3. Version Control

Monthly reports may revise prior-period data
Use content_hash to track data changes
Retain revision_flag to mark revised values

4. Parsing Priority

PDF structured parsing (priority)
HTML table parsing (fallback)
TXT fixed-width parsing (legacy) </essential_principles>

Ingest - Fetch and parse the latest or a specified month's WASDE report
Backfill - Batch backfill historical WASDE data
Validate - Verify consistency between existing data and the latest report
Configure - Configure parameters such as commodity scope and output path

Wait for a response before continuing.

After reading the workflow, follow its steps exactly.

<reference_index> Commodity Reference Documents (references/)

Document	Content
commodities-overview.md	Overview of all commodities and unit mappings
grains.md	Grains (Wheat, Corn, Rice, Barley, Sorghum, Oats)
oilseeds.md	Oilseeds (Soybeans, Soybean Oil, Soybean Meal)
cotton.md	Cotton (US & World)
livestock.md	Livestock products (Beef, Pork, Poultry, Eggs, Dairy)
sugar.md	Sugar (US & Mexico)
validation-rules.md	Validation rules and sanity checks
failure-modes.md	Failure modes and mitigation strategies
</reference_index>

<workflows_index>

Workflow	Purpose
ingest.md	Fetch and parse single or multiple WASDE reports
backfill.md	Batch backfill historical data
validate.md	Verify data consistency and revision tracking
configure.md	Configure commodity scope and output settings
</workflows_index>

<templates_index>

Template	Purpose
config.yaml	Standard configuration file template
us-balance-schema.yaml	US supply and demand table schema
world-balance-schema.yaml	World supply and demand table schema
</templates_index>

<scripts_index>

Script	Purpose
discover_releases.py	Discover WASDE release dates and URLs
fetch_report.py	Download PDF/HTML reports
parse_tables.py	Parse tables and extract data
validate_data.py	Run validation checks
</scripts_index>

<examples_index> Example Output (examples/)

Document	Content
corn_us_balance.json	Example US corn supply and demand balance sheet
wheat_world_balance.json	Example world wheat supply and demand balance sheet
ingest_status_success.json	Example status report for a successful run
backfill_result.json	Example batch backfill operation results
</examples_index>

<quick_start> CLI Quick Start:

# 抓取最新報告的所有商品
macro-skills wasde ingest --commodities=all --out=./data

# 只抓取穀物類
macro-skills wasde ingest --commodities=grains --out=./data

# 回補過去 24 個月的大豆數據
macro-skills wasde backfill --commodity=soybeans --months=24 --out=./data

# 驗證現有數據
macro-skills wasde validate --data-dir=./data

Library Quick Start:

from macro_skills.wasde import WASDEIngestor

ingestor = WASDEIngestor(
    commodities=["wheat", "corn", "soybeans"],
    scope=["us", "world"],
    output_dir="./data/wasde"
)

result = ingestor.ingest_latest()
print(result.status)

</quick_start>

<success_criteria> When the skill executes successfully:

Correctly identify commodity categories and corresponding tables
Parse supply and demand data for all specified commodities
Correctly convert and label units
Pass balance formula validation
Output standardized parquet dataset
Generate ingest_status report </success_criteria>

<detailed_usage_examples> Detailed Usage Examples

Example 1: Fetch the latest corn and soybean data

User: 幫我抓取最新 WASDE 的玉米和大豆供需數據

Assistant 執行步驟:
1. 讀取 workflows/ingest.md
2. 呼叫 discover_releases.py 找到最新發布 (2025-01-10)
3. 下載 PDF 報告
4. 解析 corn_us, corn_world, soybeans_us, soybeans_world 表格
5. 驗證平衡公式
6. 輸出 parquet 檔案
7. 回報 ingest_status

預期輸出:
- ./data/wasde/curated/corn_us_balance.parquet
- ./data/wasde/curated/corn_world_balance.parquet
- ./data/wasde/curated/soybeans_us_balance.parquet
- ./data/wasde/curated/soybeans_world_balance.parquet

Example 2: Backfill historical wheat data and track corrections

User: 回補過去 12 個月的小麥數據，我想追蹤 USDA 的修正歷程

Assistant 執行步驟:
1. 讀取 workflows/backfill.md
2. 確定日期範圍 (2024-01 至 2024-12)
3. 依序處理 12 期報告
4. 比對每期數據，標記修正值
5. 輸出包含 revision_flag 的完整歷史數據

預期輸出:
- 12 期小麥數據
- 修正追蹤報告

Example 3: Verify consistency of existing data

User: 檢查我現有的 WASDE 數據是否與最新報告一致

Assistant 執行步驟:
1. 讀取 workflows/validate.md
2. 載入現有 parquet 檔案
3. 抓取最新報告
4. 逐欄位比對
5. 標記差異與修正
6. 輸出驗證報告

預期輸出:
- validation_report.json (差異列表)
- 修正建議

</detailed_usage_examples>

<data_pipeline_architecture> Data pipeline architecture

[USDA WASDE Website]
         |
         v
+-------------------+
| discover_releases |  --> 找到 PDF URL 與發布日期
+-------------------+
         |
         v
+-------------------+
|   fetch_report    |  --> 下載 PDF (fallback: HTML)
+-------------------+
         |
         v
+-------------------+
|   parse_tables    |  --> 提取供需表格數據
+-------------------+
         |
         v
+-------------------+
|   normalize_data  |  --> 標準化欄位與單位
+-------------------+
         |
         v
+-------------------+
|   validate_data   |  --> 平衡公式與範圍檢查
+-------------------+
         |
         v
+-------------------+
|   write_output    |  --> Parquet / JSON
+-------------------+
         |
         v
[Curated Dataset]

Directory structure:

data/wasde/
├── raw/                    # 原始 PDF 檔案
│   └── 2025-01-10/
│       └── wasde0125.pdf
├── intermediate/           # 中間解析結果
│   └── 2025-01-10/
│       ├── tables.json
│       └── ingest_status.json
└── curated/                # 最終標準化數據
    ├── corn_us_balance.parquet
    ├── corn_world_balance.parquet
    ├── wheat_us_balance.parquet
    └── ...

</data_pipeline_architecture>

About this skill