instruction.md
# 멀티에셋 모닝 브리핑
## What this skill does
서울 기준 아침에 끝난 가장 최근 미국·한국 세션을 각각 확인하고, 글로벌 자산의 움직임과 한국시장 전달 경로를 근거 링크가 있는 한국어 리서치 메모로 작성한다. 조회·작성 전용이며 실시간 호가, 매매 신호, 투자 권유는 제공하지 않는다.
## When to use
- 모닝 브리핑, Morning Call, 오버나이트 시황, 데일리 마켓 브리핑 작성
- 기존 브리핑의 숫자·세션·출처·인과관계 검수
## Workflow
### 1. 기준일과 세션
`briefing-date`는 서울 기준 브리핑 날짜다. 미국 오버나잇과 한국 전일은 같은 달력 날짜라고 가정하지 않는다.
```bash
npx -y @nomadamas/k-skill@0 exec multi-asset-morning-briefing scripts/market_data.py -- snapshot --briefing-date 2026-09-17
```
`sessions.us_overnight`는 Cboe VIX 관측일, `sessions.korea_previous`는 ECOS KOSPI 관측일로 독립 확인한다. 한 원천이 실패하면 다른 시장의 결과는 유지하고 해당 세션만 `null`로 둔다. 브리핑 시점 이후 관측값, 장중 값, 다음 거래일 값은 사용하지 않는다.
개별 공식 원천을 다시 확인할 때:
```bash
npx -y @nomadamas/k-skill@0 exec multi-asset-morning-briefing scripts/market_data.py -- yields --last 5 --session 2026-09-16
npx -y @nomadamas/k-skill@0 exec multi-asset-morning-briefing scripts/market_data.py -- cboe-vix --last 5 --session 2026-09-16
npx -y @nomadamas/k-skill@0 exec multi-asset-morning-briefing scripts/market_data.py -- ecb-fx --last 5 --session 2026-09-16
```
helper의 `observed`, `session`, `basis`, `unit`, `stale`, `warnings`, `failures`를 확인한다. `stale: true`인 값은 목표 세션 값으로 쓰지 않는다.
### 2. 공식 원천 우선 조사
helper는 전체 브리핑이 아니라 검증 가능한 기초 사실 집합이다. browser/web 조사로 다음을 보강한다.
1. 중앙은행·통계기관·재무부·거래소·기업 IR/공시
2. 지수·선물·원자재 거래소 또는 신뢰할 수 있는 시세 제공처
3. Reuters, Bloomberg, CNBC, 연합인포맥스, 연합뉴스 등 원문 기사
최소 확인 범위:
- S&P 500, Nasdaq, Dow, Russell 2000, SOX, VIX와 주요 업종
- 미국 국채 2Y·10Y·30Y, 2s10s, 핵심 거시지표 실제·예상·이전치
- DXY, USD/JPY, EUR/USD, GBP/USD, USD/KRW 현물 또는 NDF
- WTI·Brent(`USD/bbl`), 금(`USD/oz`), 구리(`USD/lb` 또는 `USD/metric ton`)
- 지수 기여도가 크거나 촉매가 확인된 개별 종목
- KOSPI·KOSDAQ, 국내 3Y·10Y, EWY·야간선물·ADR 등 한국 선행 신호
가격은 `종목/계약월 · 세션 · 종가/정산가/장중 · 값 · 등락률`을 한 묶음으로 확인한다. 숫자나 방향의 차이를 세션·계약월·단위로 설명하지 못하면 제외한다.
### 3. 뉴스와 기준시각
브리핑 날짜의 `07:00 KST`를 기본 cutoff로 사용하되 사용자가 다른 작성 시각을 주면 그 시각을 따른다. 발행시각이 cutoff 이후인 기사와 장 마감 뒤 기사는 해당 브리핑 근거에서 제외한다.
제목 부분 문자열만으로 섹션을 분류하지 않는다. 기사 원문을 열어 대상 자산과 촉매를 확인한다. 예를 들어 `공원화`는 원화 기사로, 단순 달러 금액은 FX 기사로 분류하지 않는다.
### 4. 향후 5거래일 일정
Investing.com HTML 자동 파싱을 사용하지 않는다. 브리핑일부터 시작하는 **한국 거래일 5개**를 먼저 계산하고, Fed·BLS·BEA·U.S. Treasury·ECB·BOJ·한국은행·통계청·기업 IR 등 공식 일정에서 날짜와 시간을 확인한다. 공식 일정이 불명확할 때만 신뢰 가능한 경제 캘린더로 교차 확인한다.
각 일정에는 날짜, KST 시간(확인 가능할 때), 이벤트, 예상/이전치(확보 시), 시장이 주목할 이유를 적고 공식 일정 링크를 붙인다.
### 5. 분석과 출처
- 사실·수치: 원천 수치와 기준시점이 확인된 경우만 사용
- 원인: 기사나 공식 발언이 직접 설명했거나 뉴스와 가격 반응이 함께 뒷받침할 때만 단정
- 전망: `~할 경우`, `확인할 변수`, `부담 요인`처럼 조건부로 표시
- 한국시장: 업종·스타일·외국인 수급·환율·금리 경로와 반대 위험을 함께 제시
모든 사실·수치·해석 bullet 끝에 직접 근거가 되는 Markdown 링크를 하나 이상 붙인다. 검색 결과 제목이나 스니펫은 출처로 쓰지 않는다.
세부 산술·단위·문체 검수 규칙은 필요할 때 읽는다.
```bash
npx -y @nomadamas/k-skill@0 read multi-asset-morning-briefing references/format-rules.md
```
## Output contract
파일을 만들지 않고 대화에만 출력한다. 첫 줄과 7개 섹션 이름은 정확히 다음과 같다.
```text
YYYY.MM.DD Morning Market Briefing
1. Summary
2. Rates
3. FX
4. Commodity
5. Equity, Vol
6. 한국 증시
7. 주요 일정
```
- 각 항목은 `•`로 시작하고 핵심 수치/주장과 의미를 한 문장 안에서 연결한다.
- 문장 끝은 `상승`, `하락`, `확대`, `축소`, `부담`, `긍정적`, `핵심`, `변수`, `확인 필요` 같은 리서치 메모형으로 맞춘다.
- 작성자 이름과 `by` 표기를 넣지 않는다.
- 항목을 못 구했으면 그 항목만 `확보하지 못함`으로 표시한다. 근거가 전혀 없는 섹션만 `확보 실패`로 표시한다.
## Done when
- 미국·한국 세션이 독립적으로 확인됨
- 7개 섹션과 향후 5개 한국 거래일 일정이 모두 존재함
- 핵심 자산의 값·단위·세션·가격 기준이 명시됨
- 모든 사실·수치·해석에 직접 근거 링크가 있음
- 확인되지 않은 값, cutoff 이후 기사, 서술형 종결, 작성자 표기가 없음
## Failure modes
| 상황 | 동작 |
| --- | --- |
| 한 원천의 HTTP·파싱 실패 | `failures[]`에 격리하고 다른 원천·섹션 계속 진행 |
| `stale: true` 또는 세션 불일치 | 대체 원천 확인, 해결되지 않으면 값 제외 |
| 공식 일정 접근 실패 | 다른 공식 기관 일정 → 신뢰 가능한 캘린더 순으로 교차 확인 |
| 출처 간 값 불일치 | 세션·계약월·단위·종가/정산가를 대조; 해결되지 않으면 제외 |
| 핵심 근거 전체 실패 | 기억이나 추정으로 채우지 않고 확보 실패를 명시 |
README.md
# Multi-Asset Morning Briefing
> 글로벌 주식·금리·외환·원자재·변동성을 하나의 흐름으로 연결하고, 한국시장 전달 경로와 향후 5거래일 일정을 근거 링크와 함께 정리하는 AI 에이전트 스킬
[](https://www.python.org/)
[](scripts/market_data.py)
[](#공식-데이터-원천)
[](tests/test_market_data.py)
[](https://github.com/NomaDamas/k-skill/pull/675)
## 왜 만들었나
모닝 브리핑은 숫자를 많이 모으는 것보다 **같은 세션의 숫자를 정확히 연결하는 일**이 어렵습니다.
- 서울 아침 기준 미국 오버나잇과 한국 전일의 거래일이 다를 수 있음
- 휴장·주말·발표 지연 때문에 달력상 `D-1`과 실제 최근 완료 세션이 어긋남
- 현물 종가, 선물 정산가, 장중 값이 섞이면 방향과 등락률이 충돌함
- 뉴스 제목의 단순 키워드 분류는 `공원화 → 원화` 같은 오탐을 만듦
- 한 데이터 원천의 장애가 전체 브리핑 실패로 번질 수 있음
이 스킬은 미국과 한국의 세션을 **독립 판별**하고, 공식 원천의 관측일·단위·가격 기준을 보존하며, 실패한 원천만 격리합니다.
## 결과물
파일을 만들지 않고 대화에 다음 7개 섹션을 출력합니다.
```text
YYYY.MM.DD Morning Market Briefing
1. Summary
2. Rates
3. FX
4. Commodity
5. Equity, Vol
6. 한국 증시
7. 주요 일정
```
각 bullet은 핵심 수치 또는 주장을 먼저 제시하고, 의미와 한국시장 전달 경로를 한 문장 안에서 연결하며, 직접 근거가 되는 Markdown 링크로 끝납니다.
### 실제 출력 예시
아래는 `2026-09-17 15:59 KST`에 스킬을 끝까지 실행한 결과입니다. 콘텐츠 cutoff는 `07:00 KST`, 독립 확인된 미국·한국 최근 완료 세션은 모두 `2026-09-16`이며 공식 데이터 helper의 원천 실패는 0건이었습니다. helper 실행 뒤 최소 확인 범위 전체를 공식기관·거래소·주요 통신사로 보강했으며, 검증하지 못한 항목은 그대로 표시했습니다. 시점이 고정된 실행 예시이므로 최신 브리핑은 다음 명령으로 다시 생성해야 합니다.
```bash
npx -y @nomadamas/k-skill@0 exec multi-asset-morning-briefing scripts/market_data.py -- \
snapshot --briefing-date 2026-09-17
```
<details open>
<summary><strong>2026.09.17 Morning Market Briefing</strong></summary>
#### 1. Summary
• 연준이 기준금리 목표범위를 `3.75~4.00%`로 25bp 인상하고 연말 정책금리 중앙값을 `4.1%`로 제시해 시장의 중심축이 인하 기대에서 추가 긴축 위험으로 이동. [Federal Reserve Statement](https://www.federalreserve.gov/newsevents/pressreleases/monetary20260916a.htm) [Federal Reserve SEP](https://www.federalreserve.gov/monetarypolicy/fomcprojtabl20260916.htm)
• S&P 500 `-0.45%`, Dow `-1.21%`, Russell 2000 `-0.40%`로 약세였지만 Nasdaq `-0.01%`, SOX `+0.63%`로 반도체·기술주 상대강도 지속. [AP](https://apnews.com/article/b082e78c9b572b6b0a94b8033a0ee96b) [Nasdaq](https://indexes.nasdaq.com/index/Overview/SOX)
• 미 2년물 `+7bp`와 원/달러 `+9.2원`이 동시에 상승해 연준 인상 충격이 단기금리와 원화 약세로 전달, 외국인 수급에는 부담. [U.S. Treasury](https://home.treasury.gov/resource-center/data-chart-center/interest-rates/daily-treasury-rates.csv/2026/all?type=daily_treasury_yield_curve&field_tdr_date_value=2026&page&_format=csv) [서울경제](https://en.sedaily.com/finance/2026/09/16/won-slides-past-1370-as-oil-prices-push-us-yields-to-5)
• 유가는 전일 급등분을 일부 반납했지만 배럴당 `100달러`를 웃돌고 미국 수입물가는 전년 대비 `7.0%` 상승해 인플레이션 재가속 위험 잔존. [PFL Petroleum](https://pflpetroleum.com/reports/petroleum-daily-report-9-16-2026/) [BLS](https://www.bls.gov/news.release/ximpim.nr0.htm)
• KOSPI 주간시장의 `+1.37%` 반등과 달리 미국 상장 한국 ETF EWY는 오버나이트 `-0.54%`로 마감해 국내 반도체 강세와 연준 이후 해외 위험선호 간 괴리 확인 필요. [서울경제](https://dcb0ucee30q2g.cloudfront.net/finance/2026/09/16/kospi-rebounds-past-6700-despite-17-trillion-won-foreign) [EWY](https://stockanalysis.com/etf/ewy/)
#### 2. Rates
• 연준은 `12대 0` 만장일치로 25bp 인상해 연방기금금리 목표범위를 `3.75~4.00%`로 조정하고 지급준비금 이율을 `3.90%`로 인상, 긴축 재개 확정. [Federal Reserve](https://www.federalreserve.gov/newsevents/pressreleases/monetary20260916a.htm) [Implementation Note](https://www.federalreserve.gov/newsevents/pressreleases/monetary20260916a1.htm)
• 2026년 SEP 중앙값은 GDP `2.3%(6월 2.2%)`, 실업률 `4.1%(4.3%)`, PCE `3.7%(3.6%)`, Core PCE `3.4%(3.3%)`, 연말 정책금리 `4.1%(3.8%)`로 성장·물가·금리 전망 동반 상향. [Federal Reserve SEP](https://www.federalreserve.gov/monetarypolicy/fomcprojtabl20260916.htm)
• 미 국채 2년물 `4.74%(+7bp)`, 10년물 `5.01%(+1bp)`, 30년물 `5.35%(-1bp)`로 단기물 중심 Bear Flattening 발생. [U.S. Treasury](https://home.treasury.gov/resource-center/data-chart-center/interest-rates/daily-treasury-rates.csv/2026/all?type=daily_treasury_yield_curve&field_tdr_date_value=2026&page&_format=csv)
• 2s10s는 `+27bp`로 전일 `+33bp`에서 `6bp` 축소돼 추가 긴축 기대가 단기 구간에 집중, 성장주 할인율과 레버리지 자산에 부담. [U.S. Treasury](https://home.treasury.gov/resource-center/data-chart-center/interest-rates/daily-treasury-rates.csv/2026/all?type=daily_treasury_yield_curve&field_tdr_date_value=2026&page&_format=csv)
• 미국 8월 수입물가 `+0.7% MoM·+7.0% YoY`, 수출물가 `+0.6% MoM·+8.6% YoY`, 비연료 수입물가 `+0.8% MoM`로 관세·상품가격발 물가 압력이 연준의 추가 긴축 논리를 지지. [BLS](https://www.bls.gov/news.release/ximpim.nr0.htm)
#### 3. FX
• DXY 보조 종가는 `99.68(+0.03%)`로 상승했으며 ICE 공식 정산값이 아닌 Yahoo Finance 기반 집계값이므로 방향 확인용 SECONDARY. [Aegean DXY](https://aegeanintel.com/data/series/dxy/)
• ECB 기준 EUR/USD `1.1537(-0.02%)`, GBP/USD `1.34558(-0.20%)`로 하락해 유럽통화 대비 달러 우위 확인. [ECB](https://data-api.ecb.europa.eu/service/data/EXR/D.USD+JPY+GBP.EUR.SP00.A?format=csvdata&startPeriod=2026-09-03&endPeriod=2026-09-17)
• USD/JPY는 `155.049(+0.03%)`로 상승해 엔화 약세가 이어졌으며 일본 수입물가와 아시아 통화 변동성에는 부담. [ECB](https://data-api.ecb.europa.eu/service/data/EXR/D.USD+JPY+GBP.EUR.SP00.A?format=csvdata&startPeriod=2026-09-03&endPeriod=2026-09-17)
• 서울 현물 원/달러 종가는 `1,368.6원`, 전일 대비 `+9.2원(+0.68%)`으로 원화 약세 심화, 외국인 현물 수급에는 부담이나 수출주 환산 실적에는 완충 변수. [서울경제](https://en.sedaily.com/finance/2026/09/16/won-slides-past-1370-as-oil-prices-push-us-yields-to-5)
• 확인 가능한 동일 기준시점의 원/달러 1개월 NDF 종가를 확보하지 못해 서울 현물 종가와 오버나이트 환율을 혼합하지 않음. [한국은행 ECOS](https://ecos.bok.or.kr/)
#### 4. Commodity
• WTI 10월물은 `102.43달러/배럴(-3.21%)`, Brent 11월물은 `105.83달러/배럴(-2.69%)`에 정산돼 전일 급등분 일부 반납, 정유·에너지에는 차익실현 부담이고 항공·운송에는 비용 압력 완화. [PFL Petroleum](https://pflpetroleum.com/reports/petroleum-daily-report-9-16-2026/) [Gate](https://www.gate.com/en-us/news/detail/wti-crude-falls-340-to-10243-brent-drops-269-on-september-16-17873479)
• 유가 하락 배경은 사우디의 추가 아시아 공급과 수출 차질 완화 기대였지만 월간 상승률은 여전히 `16% 이상`으로 국내 물가·항공·화학 마진의 불확실성 지속. [PFL Petroleum](https://pflpetroleum.com/reports/petroleum-daily-report-9-16-2026/)
• COMEX 금 12월물은 `4,346.30달러/온스(+1.27%)`에 정산돼 연준 결정 전후 안전자산 수요 유지, 다만 단일 2차 원천 기준 SECONDARY. [Indy Finance](https://www.indy.finance/news/gold-settles-at-4-346-30-as-traders-brace-for-fed-rate-decision)
• LME 구리 Cash Settlement는 `14,227달러/톤(+1.30%)`, 3개월물은 `14,232달러/톤(+1.10%)`, 재고는 `254,150톤(+1.98%)`으로 가격과 재고가 동반 상승해 수요 신호로 단정하기보다 공급·재고 흐름 확인 필요. [Westmetall LME data](https://www.westmetall.com/en/markdaten.php/leistungen/en/en/westmetall.html?action=table&field=LME_Cu_cash)
#### 5. Equity, Vol
• S&P 500 `7,551.81(-0.45%)`, Dow `51,461.90(-1.21%)`, Nasdaq `25,978.42(-0.01%)`, Russell 2000 `2,858.81(-0.40%)`로 대형 기술주 대비 금융·경기민감주 상대 약세. [AP](https://apnews.com/article/b082e78c9b572b6b0a94b8033a0ee96b)
• VIX는 `17.71pt`, 전일 `17.20pt` 대비 `+0.51pt` 상승해 지수 낙폭보다 헤지 수요 증가가 두드러진 세션. [Cboe](https://cdn.cboe.com/api/global/us_indices/daily_prices/VIX_History.csv)
• SOX는 `11,246.11(+0.63%)`로 상승했고 Nvidia `+0.8%`, AMD `+1.6%`로 반도체 바스켓이 연준 인상에도 상대강도 유지. [Nasdaq](https://indexes.nasdaq.com/index/Overview/SOX) [AP](https://apnews.com/article/e2e82957e490b7be205db6013f621c3d)
• S&P 500 11개 업종 중 기술주만 `+0.12%`로 상승하고 10개 업종이 하락해 지수 보합에 가까운 Nasdaq과 달리 시장 폭은 취약. [WMTMT market note](https://www.wmtmt.com/notes/daily-2026-09-16)
• Huntington Bancshares `-5.6%`, Citizens Financial `-4.8%`, JPMorgan `-1.0%`로 은행주가 하락해 금리 인상이 금융주 전반의 즉각적 호재로 연결되지 않음. [AP](https://apnews.com/article/e2e82957e490b7be205db6013f621c3d)
• J.B. Hunt `-13.3%`는 3분기 이익이 2분기 대비 `5~10%` 감소할 수 있다는 비용 경고에 따른 개별 실적 충격, 운송주 비용 압력 확인 필요. [AP](https://apnews.com/article/e2e82957e490b7be205db6013f621c3d)
#### 6. 한국 증시
• KOSPI `6,717.97(+1.37%)`, KOSDAQ `815.98(+0.44%)`로 동반 상승해 KOSPI는 4거래일 연속 하락 뒤 반등. [한국은행 ECOS](https://ecos.bok.or.kr/api/StatisticSearch/sample/json/kr/1/10/802Y001/D/20260916/20260916) [서울경제](https://dcb0ucee30q2g.cloudfront.net/finance/2026/09/16/kospi-rebounds-past-6700-despite-17-trillion-won-foreign)
• 삼성전자 `+2.01%`, SK하이닉스 `+4.08%`로 대형 반도체가 반등을 주도했고 미국 SOX `+0.63%`가 오버나이트 상대강도 신호, 장중 외국인 수급이 지속성의 핵심 변수. [서울경제](https://dcb0ucee30q2g.cloudfront.net/finance/2026/09/16/kospi-rebounds-past-6700-despite-17-trillion-won-foreign) [Nasdaq SOX](https://indexes.nasdaq.com/index/Overview/SOX)
• KOSPI에서 기관이 약 `1.21조원` 순매수했지만 외국인은 약 `1.68조원`, 개인은 약 `1.18조원` 순매도해 지수 반등과 수급의 괴리 확대. [서울경제](https://dcb0ucee30q2g.cloudfront.net/finance/2026/09/16/kospi-rebounds-past-6700-despite-17-trillion-won-foreign)
• 미국 상장 한국 ETF EWY는 `175.54달러(-0.54%)`로 마감해 국내 현물 반등을 이어가지 못했고, 연준 인상·원화 약세를 반영한 해외 투자심리는 부담. [EWY](https://stockanalysis.com/etf/ewy/)
• 원/달러 현물 `1,368.6원(+9.2원)`과 외국인 대규모 순매도가 동반돼 환율 안정 전까지 외국인 주도 추세 전환 판단은 유보. [서울경제 FX](https://en.sedaily.com/finance/2026/09/16/won-slides-past-1370-as-oil-prices-push-us-yields-to-5) [연합뉴스](https://en.yna.co.kr/view/AEN20260916008600320)
• 동일 세션 국내 국고채 3년·10년 종가, KOSPI200 야간선물, 주요 한국 ADR의 일관된 종가를 신뢰 가능한 원천에서 교차 확인하지 못해 해당 수치는 확보하지 못함. [금융투자협회 채권정보센터](https://www.kofiabond.or.kr/)
#### 7. 주요 일정
• 향후 한국 거래일 5개는 `09.17·09.18·09.21·09.22·09.23`, 일정 시각은 모두 KST 기준. [한국은행](https://www.bok.or.kr/eng/stats/statsPublictSchdul/listCldr.do?date=25135-09&menuNo=400359) [New York Fed](https://www.newyorkfed.org/research/calendars/i-sep26.html)
• `09.17 12:00 KST` 한국은행 2026년 상반기 지식재산권 무역수지 잠정치는 서비스수지와 원화 펀더멘털 점검 변수. [한국은행](https://www.bok.or.kr/eng/stats/statsPublictSchdul/listCldr.do?date=25135-09&menuNo=400359)
• `09.17 21:30 KST(08:30 ET)` 미국 신규 실업수당 청구건수는 컨센서스 `207K`, 이전 `206K`; 같은 시각 8월 주택착공은 `1.30M` 예상, 이전 `1.239M`으로 연준 인상 직후 고용·주택 민감도 확인 핵심. [New York Fed Calendar](https://www.newyorkfed.org/research/calendars/i-sep26.html) [Census](https://www.census.gov/construction/soc/schedule.html) [WSJ survey via MarketScreener](https://www.marketscreener.com/news/housing-starts-on-tap-data-week-ahead-ce785bd2df8af126)
• `09.17 21:30 KST` 필라델피아 연은 제조업지수는 `34.0` 예상, 이전 `47.4`; `23:00 KST` 잠정주택판매는 `+0.5%` 예상, 이전 `-2.3%`로 성장 모멘텀 교차 확인 변수. [New York Fed Calendar](https://www.newyorkfed.org/research/calendars/i-sep26.html) [WSJ survey via MarketScreener](https://www.marketscreener.com/news/housing-starts-on-tap-data-week-ahead-ce785bd2df8af126)
• `09.18 06:00 KST` 한국 8월 생산자물가지수, `12:00 KST` 2025년 공공부문계정 발표로 국내 물가·재정 흐름 점검 핵심. [한국은행](https://www.bok.or.kr/eng/stats/statsPublictSchdul/listCldr.do?date=25135-09&menuNo=400359)
• `09.18 22:15 KST(09:15 ET)` 미국 8월 산업생산은 `+0.3%` 예상, 이전 `+0.2%`; 설비가동률은 `76.4%` 예상, 이전 `76.3%`로 경기 강도와 추가 긴축 여력 판단 변수. [Federal Reserve G.17](https://www.federalreserve.gov/releases/g17/) [WSJ survey via MarketScreener](https://www.marketscreener.com/news/housing-starts-on-tap-data-week-ahead-ce785bd2df8af126)
• `09.18 23:00 KST(10:00 ET)` 미국 8월 주별 고용·실업 발표로 지역별 노동시장 확산 정도 확인 필요. [BLS](https://www.bls.gov/schedule/2026/)
• `09.21 12:00 KST` 한국은행 2024년 연장표 산업연관표 발표로 반도체·에너지 가격 충격의 업종별 투입 경로 확인 필요. [한국은행](https://www.bok.or.kr/eng/stats/statsPublictSchdul/listCldr.do?date=25135-09&menuNo=400359)
• `09.22 23:00 KST(10:00 ET)` 미국 9월 리치먼드 연은 제조업지수 발표, 이전 `4`로 지역 제조업 확장 지속 여부가 경기·금리의 확인 변수. [New York Fed Calendar](https://www.newyorkfed.org/research/calendars/i-sep26.html) [Richmond Fed series](https://www.investing.com/economic-calendar/richmond-manufacturing-index-263)
• `09.23 06:00 KST` 한국 9월 소비자동향조사 발표, 8월 소비자심리지수 `104.5`에서의 반등 여부가 내수·소비 업종의 핵심 변수. [한국은행](https://www.bok.or.kr/eng/stats/statsPublictSchdul/listCldr.do?date=25135-09&menuNo=400359) [이전치](https://tradingeconomics.com/south-korea/consumer-confidence)
</details>
## 처리 흐름
```mermaid
flowchart LR
A[서울 기준 브리핑일] --> B1[Cboe VIX 관측일]
A --> B2[한국은행 ECOS 관측일]
B1 --> C1[미국 최근 완료 세션]
B2 --> C2[한국 최근 완료 세션]
C1 --> D[공식 기초 데이터]
C2 --> D
D --> E[원문 뉴스·공식 일정 보강]
E --> F[세션·단위·cutoff 검증]
F --> G[7개 섹션 브리핑]
```
## 공식 데이터 원천
| 원천 | 수집 항목 | 검증 기준 |
|---|---|---|
| [U.S. Treasury](https://home.treasury.gov/resource-center/data-chart-center/interest-rates) | 미국 국채 2Y·10Y·30Y, 2s10s | 미국 목표 세션과 동일한 관측일 |
| [Cboe](https://www.cboe.com/tradable_products/vix/) | VIX 종가, 미국 최근 완료 세션 | `pt`, regular close |
| [ECB Data Portal](https://data.ecb.europa.eu/data/datasets/EXR) | EUR/USD·USD/JPY·GBP/USD | reference rate, 통화쌍별 단위 |
| [한국은행 ECOS](https://ecos.bok.or.kr/) | KOSPI·KOSDAQ, 한국 최근 완료 세션 | `pt`, 한국시장 종가 |
로그인·API 키·프록시 없이 공개 read-only endpoint를 직접 호출합니다. FRED 시계열은 발표 지연을 명시하는 보조 fallback으로만 사용하며, 목표 세션과 관측일이 다르면 브리핑에서 제외합니다.
## 빠른 시작
### 1. 스킬 설치·갱신
```bash
npx -y @nomadamas/k-skill@0 update
```
### 2. 에이전트 지침 확인
```bash
npx -y @nomadamas/k-skill@0 instruct multi-asset-morning-briefing
```
### 3. 공식 기초 데이터 수집
```bash
npx -y @nomadamas/k-skill@0 exec multi-asset-morning-briefing scripts/market_data.py -- \
snapshot --briefing-date 2026-09-17
```
개별 원천을 다시 확인할 수도 있습니다.
```bash
# 미국 국채 금리
npx -y @nomadamas/k-skill@0 exec multi-asset-morning-briefing scripts/market_data.py -- \
yields --last 5 --session 2026-09-16
# Cboe VIX
npx -y @nomadamas/k-skill@0 exec multi-asset-morning-briefing scripts/market_data.py -- \
cboe-vix --last 5 --session 2026-09-16
# ECB 주요 통화쌍
npx -y @nomadamas/k-skill@0 exec multi-asset-morning-briefing scripts/market_data.py -- \
ecb-fx --last 5 --session 2026-09-16
```
## 신뢰성 설계
- **독립 세션 판별** — 미국은 Cboe, 한국은 ECOS의 실제 관측일로 각각 결정
- **Fail-Closed** — 세션·단위·가격 기준을 해결하지 못한 값은 추정하지 않고 제외
- **실패 격리** — 한 원천의 HTTP·파싱 실패를 `failures[]`에 기록하고 나머지 수집 지속
- **응답 크기 제한** — `Content-Length`와 스트리밍 양쪽에서 8 MiB 상한 적용
- **뉴스 cutoff** — 기본 `07:00 KST` 이후 기사를 해당 브리핑 근거에서 제외
- **공식 일정 우선** — 제3자 HTML 자동 파싱 대신 중앙은행·통계기관·기업 IR 일정 사용
## 테스트
```bash
python -m unittest discover -s multi-asset-morning-briefing/tests -p "test_*.py"
ruff format --check multi-asset-morning-briefing/scripts/market_data.py \
multi-asset-morning-briefing/tests/test_market_data.py
ruff check multi-asset-morning-briefing/scripts/market_data.py \
multi-asset-morning-briefing/tests/test_market_data.py
```
오프라인 테스트 30건이 다음을 검증합니다.
- 공식 CSV·JSON 파싱과 FRED ZIP 병합
- 미국·한국 휴장일이 다른 경우의 독립 세션
- 2s10s·bp 변화·curve 분류 산술
- 미국 국채·FX·VIX의 자산별 단위
- 비정상 JSON과 원천별 장애의 구조화된 실패
- 선언/스트리밍 응답 크기 제한
- 완성 브리핑의 제목·7개 섹션·링크·리서치 메모형 문체
## 저장소 구조
```text
multi-asset-morning-briefing/
├── skill.json # frontmatter·profile 원본
├── instruction.md # 에이전트 workflow 원본
├── SKILL.md # 자동 생성 CLI adapter
├── scripts/market_data.py # stdlib 공식 데이터 helper
├── references/format-rules.md # 산술·단위·문체 검수 규칙
└── tests/ # 오프라인 회귀·E2E 계약 테스트
```
`SKILL.md`는 직접 편집하지 않으며 `skill.json`과 `instruction.md`에서 생성합니다.
## 기여 현황
- 제안 이슈: [NomaDamas/k-skill #674](https://github.com/NomaDamas/k-skill/issues/674)
- 업스트림 PR: [NomaDamas/k-skill #675](https://github.com/NomaDamas/k-skill/pull/675)
- 라이선스: [MIT](https://github.com/NomaDamas/k-skill/blob/dev/LICENSE)
references/format-rules.md
# 모닝 브리핑 작성·검증 규칙
`multi-asset-morning-briefing`이 본문을 쓰기 전에 따르는 세부 규칙이다. 우선순위는 **정확성 > 출처 품질 > 관련성 > 간결성 > 서사**다.
작성 순서는 **Data first. Interpretation second. Outlook last.** 이고, 확인되지 않은 정보로 문서를 채우지 않는다.
## 1. 기준시점과 세션 (Temporal rules)
브리핑 기준일 `D`는 서울시간 오늘이고 제목 날짜도 `D`다.
- 미국 오버나잇과 한국 전일은 공식 관측일로 각각 확인한다. 달력상 `D-1`을 공통 적용하지 않는다.
- 미국 휴장과 한국 휴장이 다르면 두 세션 날짜가 달라도 정상이다.
- 일요일·월요일에는 토요일을 세션으로 만들지 않고 각 시장의 직전 완료 거래일을 사용한다.
수집한 값의 **관측일이 목표 세션과 같은지 먼저 확인**한다. 공식 집계 사이트는 시리즈별 발표 지연이 있어 FX·원자재·수급 데이터가 며칠 뒤처질 수 있다. 관측일이 다르면 그 값을 해당 세션 값처럼 쓰지 않는다.
자산군별 기준:
| 자산군 | 기준 |
| --- | --- |
| 미국 주식 | 정규장 종가. regular close와 after-hours를 섞지 않는다. ADR도 같은 미국 정규장 기준 |
| 미국 금리 | 2Y와 10Y를 **같은 공급자·같은 기준시점**으로 맞춘다. 서로 다른 공급자나 다른 시각으로 curve를 만들지 않는다. Treasury CMT와 market closing yield를 같은 데이터로 취급하지 않는다 |
| FX | DXY와 G10 FX는 같은 NY 기준시점. USD/KRW는 서울 현물 종가와 1M NDF를 구분하고 같은 시점처럼 쓰지 않는다 |
| 원자재 | WTI와 Brent를 같은 session의 settlement 또는 같은 성격의 close로 맞춘다. intraday high/low와 settlement를 구분한다 |
| 한국 | KOSPI·KOSDAQ·국내 채권은 서울시장 종가. KOSPI200 야간선물은 야간 session임을 밝힌다 |
미국 현물시장이 휴장한 날에는 현물 지수 종가 대신 선물이나 다른 자산군을 쓰고 휴장 사실을 명시한다. 휴장일 데이터를 정규장 종가처럼 표기하지 않는다.
## 2. 출처 정책 (Source policy)
데이터 종류별로 가장 적합한 원천을 쓴다. 모든 데이터에 같은 서열을 적용하지 않는다.
**가격 데이터**
1. 해당 거래소·공식 benchmark·시장 원천
2. 신뢰 가능한 전문 market data
3. 신뢰 가능한 데이터 집계 서비스
4. 증권사 시황
5. Telegram / Blog
Telegram이나 Blog만으로 핵심 가격을 확정하지 않는다.
**경제지표·중앙은행**
1. 공식 발표기관 2. 중앙은행·정부기관 3. Reuters 등 주요 통신사 4. 경제 캘린더 서비스
예: Fed → Federal Reserve, US CPI → BLS, 한국은행 데이터 → BOK, 한국 물가 → 통계청.
**뉴스·이벤트**
1. Reuters 2. AP 등 주요 통신사 3. 공식 발표 4. 연합뉴스·연합인포맥스 5. 증권사 시황
Telegram은 discovery와 interpretation 용도로만 쓴다.
**한국**
KRX, 한국은행, 금융투자협회, 연합인포맥스, 서울 외환시장 전문 데이터를 우선한다.
**접근 실패 처리**
- 실시간 시세가 유료·제한되어 지연 호가나 집계값만 확보되면 `SECONDARY` 이하로 처리한다.
- JS 챌린지·로그인 벽이 있는 사이트는 조회 실패로 처리하고, 공식 정적 endpoint → 브라우저 런타임 → 웹 검색 순으로 대체한다.
- 조회 수단이 전혀 없으면 브리핑을 시작하지 않고 확보 실패를 보고한다. 학습 시점 기억이나 이전 브리핑으로 숫자를 채우지 않는다.
- 각 bullet의 직접 링크와 문장 안의 기준시점으로 출처를 역추적할 수 있게 한다.
## 3. 검증상태 판정
- `VERIFIED` — 공식·1차 원천에서 직접 확인했거나, 서로 독립인 원천 2개 이상에서 일치
- `SECONDARY` — 단일 2차 원천만 확보
- `CONFLICT` — 독립 원천 간 숫자 불일치
- `UNVERIFIED` — Telegram·Blog 등 비공식 원천만 확보했거나 기준시점을 확인할 수 없음
**독립 원천**의 정의: 같은 데이터 벤더나 같은 원본을 재배포한 관계가 아닌 원천. 동일 원본을 재배포한 두 매체는 1개 원천으로 센다.
집계 서비스 값을 공식 원천과 대조해 일치시키면 `VERIFIED`로 올릴 수 있다.
`CONFLICT`·`UNVERIFIED` 핵심 숫자는 최종 출력에 쓰지 않는다. `SECONDARY`는 중요도가 낮고 다른 원천이 없을 때만 제한적으로 쓴다.
## 4. 산술 검증 (Arithmetic)
모든 계산값은 원자료에서 다시 계산한다.
- `2s10s = 10Y − 2Y`
- `Δ2s10s = Δ10Y − Δ2Y`
- 반올림된 변화폭으로 정밀한 spread 변화를 만들지 않는다. 원자료가 있고 소수 1자리까지 보고할 때만 `−1.4bp` 같은 표현을 쓴다. 표시값이 `2Y +4bp`, `10Y +3bp`뿐이면 `약 1bp 축소` 정도로 쓴다.
- 상대성과는 `%p`로 쓴다. 예: `EWY −6.62%`, `EEM −2.73%` → 상대성과 `−3.89%p`.
- `%`와 `%p`를 구분하고, `bp`와 `%p`를 혼용하지 않는다.
- VIX는 포인트(pt)로 표기하고 `%`로 쓰지 않는다.
## 5. Curve 분류
같은 기준시점의 2Y·10Y 변화폭이 있을 때만 적용한다.
| 2Y | 10Y | 판정 |
| --- | --- | --- |
| ↑ | ↑ | 2Y 상승폭 > 10Y → Bear Flattening / 10Y 상승폭 > 2Y → Bear Steepening |
| ↓ | ↓ | 2Y 하락폭 > 10Y → Bull Steepening / 10Y 하락폭 > 2Y → Bull Flattening |
| ↑ | ↓ (또는 반대) | 네 가지를 억지로 적용하지 않고 `2s10s가 XXbp 확대/축소`로 직접 쓴다 |
| 변화 없음 | 변화 없음 | `2s10s 변화 없음` |
금리 방향을 확인할 수 있는데 `curve flattening` 같은 표현만 쓰지 않는다.
## 6. 섹션 작성 규칙
**주요 일정**
- 브리핑일부터 향후 한국 거래일 5개를 날짜순으로 다룬다.
- 모든 시각은 KST로 표기하고 원본 발표 시각을 괄호로 병기한다.
- 제3자 캘린더 HTML을 자동 파싱하지 않고 공식 기관 발표 일정을 우선한다.
- 발표가 끝난 항목은 결과값, 예정 항목은 예정 시각과 전월치·컨센서스를 쓴다. 발표 완료 항목을 예정처럼 쓰지 않는다.
- Stage 2 필터를 통과한 항목만 쓴다. 자리 채우기용 이벤트를 넣지 않는다.
**Summary**
본문 전체의 축약본이 아니다. 시장 전체를 설명하는 3~5개 핵심 narrative를 5~6개 bullet 이내로 정리한다. 포함 우선순위는 미국 주요 지수 결과 → 핵심 driver → 가장 큰 cross-asset move → 주요 sector rotation → 다음 정책·이벤트 리스크다. 세부 숫자는 각 섹션에서 설명한다.
**Rates**
데이터 밀도를 가장 높게 가져간다. 순서는 **Level → Change → Curve → Driver → Event**.
권장 형태:
```
• 미 국채 2년물 4.670%(+2.0bp), 10년물 5.000%(+3.0bp). 2s10s는 33.0bp로 1.0bp 확대되며 Bear Steepening
```
Fed pricing이 중요한 날에는 실제 확률을 쓰고, 국채 입찰이 중요하면 stop yield·tail/through·bid-to-cover·indirect bidder 중 검증된 수치만 쓴다. 입찰이 아직 실시되지 않았으면 결과처럼 쓰지 않는다.
**FX**
순서는 **Level → Change → Driver**. 방향만 쓰지 않는다.
- 나쁜 예: `달러 강세`
- 좋은 예: `DXY 99.48(+0.36%)로 상승`
여러 EM 통화를 확인하지 않았으면 `신흥국 통화 전반 약세`로 일반화하지 않는다.
**Commodity**
순서는 **Settlement → Daily Change → Intraday → Driver**. 예: `WTI $63.66(+0.9%), Brent $67.88(+0.2%)`. 당일 핵심 narrative와 관련 없는 상품은 생략한다. 모든 상품을 매일 채울 필요는 없다.
**Equity, Vol**
순서는 **Index → Sector → Theme Basket → Individual Stocks**. 주요 지수를 먼저 제시하고 sector·theme을 설명한 뒤, 같은 driver로 움직인 종목을 basket으로 묶는다. 종목 나열 자체를 목적으로 하지 않는다.
**한국 증시**
미국시장 내용을 그대로 반복하지 않는다. 한국 관련 overnight proxy(SOX, SK하이닉스 ADR, EWY, KOSPI200 야간선물, USD/KRW, 1M NDF, KTB 3Y·10Y)를 먼저 제시하고 그 뒤에 해석을 붙인다. USD/KRW는 서울 현물시장 종가 기준을 우선하고, KTB는 level과 일간 bp 변화로 쓴다. 미국 업종 상승을 근거로 한국 동일 업종 상승을 단정하지 않는다. `상대강세 여부 확인 가능` 수준을 넘는 전망은 추가 근거가 있을 때만 쓴다.
## 7. 한국 전이 채널
연결고리가 약하면 한국시장 섹션에 억지로 넣지 않는다.
- 반도체: SOX·Memory·AI Hardware → 삼성전자·SK하이닉스·국내 반도체 밸류체인
- 금리: UST Yield → 성장주 valuation·외국인 수급·국내 금리
- 환율: DXY·USDKRW → 외국인 수급·원화 민감 업종
- 유가: WTI·Brent → 국내 물가·정유·화학·항공·운송
- 중국: 중국 실물지표·CNH → 반도체·화학·철강·기계·원화
## 8. 금지 항목
1. 출처를 확인하지 못한 핵심 숫자를 쓴다.
2. 다른 사람이 작성한 브리핑을 데이터 원천으로 쓴다.
3. Telegram 숫자를 독립 검증 없이 쓴다.
4. 서로 다른 시점의 2Y와 10Y로 curve를 계산한다.
5. 서로 다른 시장 관례를 하나의 종가처럼 섞는다.
6. 반올림된 값으로 정밀한 spread 변화를 만든다.
7. 확인할 수 없는 숫자를 추정한다.
8. 경제일정을 채우려고 중요도 낮은 이벤트를 넣는다.
9. Rates를 정성적 설명만으로 쓴다.
10. 숫자 없는 방향성 표현(`달러 강세`, `유가 상승`, `반도체 약세`)만 쓴다.
11. curve 방향을 판정할 수 있는데 Bull/Bear를 생략한다.
12. 동시 움직임만으로 인과를 확정한다.
13. 미국시장 움직임을 한국시장 전망으로 기계적으로 변환한다.
14. Fact와 Outlook을 섞는다.
15. Summary와 본문에서 같은 내용을 반복한다.
16. 본문 내용을 마지막 체크리스트로 다시 반복한다.
17. 특정 증권사나 Telegram 채널의 문장 구조를 모방한다.
18. `주목할 필요`, `확인할 필요`, `경계감 확대` 같은 추상문장을 반복한다.
19. 데이터가 없는 자산에 내용을 억지로 만든다.
20. 출처가 충돌하는 숫자 중 원하는 값을 임의로 고른다.
21. 오류 가능성이 높은 정보를 문서 길이를 채우려고 유지한다.
22. 뉴스 제목만 보고 실제 가격 반응을 확인하지 않은 채 핵심 driver로 정한다.
23. market pricing과 확정된 정책결정을 같게 표현한다.
24. `AI 투자 중단`, `긴축 확정`처럼 근거보다 강한 표현을 쓴다.
25. 없거나 확인되지 않은 값을 그럴듯한 수치로 메운다.
## 9. 문체
기관투자가용 모닝 브리핑 문체로 쓴다.
- 각 bullet은 가능한 한 1~2문장
- 가격과 숫자를 문장 앞부분에, 해석을 숫자 뒤에
- 불필요한 접속어·같은 형용사 반복 금지
- 장황한 macro 설명, 논문식 문체, 과도한 결론문 금지
- 영어 금융용어는 실무에서 통용되는 경우 사용 가능
- 종목·시장 움직임은 구체적으로 쓴다
- 모든 bullet 끝에는 직접 근거가 되는 Markdown 링크를 붙인다.
- `~이다`, `~했다`, `~합니다`, `~보인다` 대신 명사형·요약형 종결을 사용한다.
좋은 예:
```
• 미 2년물 4.670%(+2.0bp), 10년물 5.000%(+3.0bp). 단기물 대비 장기물 상승폭이 커 2s10s는 33.0bp로 1.0bp 확대, Bear Steepening
```
나쁜 예:
```
• 다양한 거시경제적 불확실성이 복합적으로 작용하면서 시장 참가자들의 향후 정책경로에 대한 경계감이 확대되는 모습
```
## 10. 검증 실패 시 행동
필요한 데이터를 찾지 못하거나 검증에 실패하면 다음 순서를 따른다.
1. 다른 신뢰 가능한 원천에서 다시 확인한다.
2. 숫자 없이 사실 수준으로 표현할 수 있는지 판단한다.
3. 핵심 내용이 아니면 해당 bullet을 삭제한다.
4. 핵심 내용인데 검증되지 않으면 확인되지 않았음을 명시한다.
절대 추정값을 만들어 넣지 않는다.
## 11. 완료 조건
아래를 모두 만족해야 작성을 완료한다.
- 핵심 시장 숫자가 검증됨
- 기준시점이 일관됨
- Rates curve 계산이 검증됨
- 주요 driver가 실제 가격 움직임과 연결됨
- 한국시장 전이가 과도하게 추정되지 않음
- 중복·추상문장이 제거됨
- 출처를 역추적할 수 있음
**중요한 숫자는 정확하게, 중요한 변화는 명확하게, 해석은 필요한 만큼만.**
scripts/market_data.py
#!/usr/bin/env python3
"""멀티에셋 모닝 브리핑용 공개 시장 데이터 helper.
미국 재무부·Cboe·ECB·한국은행 ECOS의 공개 read-only 원천을 우선한다.
API 키·프록시·로그인이 필요 없으므로 사용자 머신에서 직접 호출한다.
핵심 원칙:
- 관측일(observation date)을 항상 함께 보고한다. 목표 세션보다 오래된 값은 stale로 표시한다.
- 결측은 보간하지 않는다. 없는 값은 None으로 남긴다.
- bp 변화는 원자료로 계산한다. 반올림된 표시값으로 다시 계산하지 않는다.
"""
import argparse
import csv
import io
import json
import re
import sys
import urllib.error
import urllib.request
import zipfile
from datetime import date, datetime, timedelta, timezone
DEFAULT_TIMEOUT = 30
FETCH_ATTEMPTS = 2
MAX_RESPONSE_BYTES = 8 * 1024 * 1024
TIMEOUT_HINT = (
"네트워크가 느리거나 차단된 환경일 수 있습니다. --timeout을 늘리거나, "
"확보 실패로 처리하고 해당 자산군을 본문에서 빼세요."
)
USER_AGENT = "multi-asset-morning-briefing/1.0 (+https://github.com/NomaDamas/k-skill)"
TREASURY_URL = (
"https://home.treasury.gov/resource-center/data-chart-center/interest-rates/"
"daily-treasury-rates.csv/{year}/all?type=daily_treasury_yield_curve"
"&field_tdr_date_value={year}&page&_format=csv"
)
FRED_URL = "https://fred.stlouisfed.org/graph/fredgraph.csv?id={ids}"
CBOE_VIX_URL = "https://cdn.cboe.com/api/global/us_indices/daily_prices/VIX_History.csv"
ECB_FX_URL = (
"https://data-api.ecb.europa.eu/service/data/EXR/"
"D.USD+JPY+GBP.EUR.SP00.A?format=csvdata&startPeriod={start}&endPeriod={end}"
)
ECOS_URL = (
"https://ecos.bok.or.kr/api/StatisticSearch/sample/json/kr/1/10/"
"802Y001/D/{start}/{end}"
)
TREASURY_COLUMNS = (("2 Yr", "y2"), ("10 Yr", "y10"), ("30 Yr", "y30"))
CURVE_EPSILON_BP = 0.05
class DataError(Exception):
"""수집·파싱 실패. code/message/source를 오류 봉투에 담는다."""
def __init__(self, code, message, source):
super().__init__(message)
self.code = code
self.message = message
self.source = source
# --------------------------------------------------------------------------
# 네트워크
# --------------------------------------------------------------------------
def fetch_bytes(url, timeout=DEFAULT_TIMEOUT):
"""URL 바이트를 가져온다. 테스트는 이 함수를 대체한다."""
request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
with urllib.request.urlopen(request, timeout=timeout) as response:
declared = response.headers.get("Content-Length")
if declared and int(declared) > MAX_RESPONSE_BYTES:
raise DataError("LIMIT", "원격 응답이 허용 크기를 초과했습니다", "http")
chunks = []
total = 0
while True:
chunk = response.read(min(64 * 1024, MAX_RESPONSE_BYTES - total + 1))
if not chunk:
break
total += len(chunk)
if total > MAX_RESPONSE_BYTES:
raise DataError("LIMIT", "원격 응답이 허용 크기를 초과했습니다", "http")
chunks.append(chunk)
return b"".join(chunks)
def fetch_with_retry(url, source, timeout=DEFAULT_TIMEOUT, attempts=FETCH_ATTEMPTS):
last_error = None
for _ in range(max(1, attempts)):
try:
return fetch_bytes(url, timeout)
except urllib.error.HTTPError as error:
last_error = DataError(
"HTTP", "HTTP {} 응답을 받았습니다: {}".format(error.code, url), source
)
except urllib.error.URLError as error:
if "timed out" in str(error.reason).lower():
last_error = DataError(
"TIMEOUT", _timeout_message(url, error.reason), source
)
else:
last_error = DataError(
"HTTP", "{}: {}".format(url, error.reason), source
)
except TimeoutError as error:
last_error = DataError("TIMEOUT", _timeout_message(url, error), source)
except OSError as error:
last_error = DataError("HTTP", "{}: {}".format(url, error), source)
raise last_error
def _timeout_message(url, detail):
return "{}: {} {}".format(url, detail, TIMEOUT_HINT)
# --------------------------------------------------------------------------
# 파싱
# --------------------------------------------------------------------------
def _parse_us_date(raw):
"""MM/DD/YYYY -> YYYY-MM-DD."""
return datetime.strptime(raw.strip(), "%m/%d/%Y").date().isoformat()
def _parse_iso(raw):
return date.fromisoformat(raw.strip())
def _to_float(raw):
cell = raw.strip()
if not cell or cell in {".", "N/A", "NA", "null"}:
return None
try:
return float(cell.replace(",", ""))
except ValueError:
return None
def parse_treasury_csv(text):
"""재무부 CSV 텍스트 -> 날짜 오름차순 레코드 목록."""
reader = csv.reader(io.StringIO(text))
try:
header = [column.strip() for column in next(reader)]
except StopIteration:
raise DataError("PARSE", "재무부 CSV가 비어 있습니다", "us-treasury")
index = {name: position for position, name in enumerate(header)}
if "Date" not in index:
raise DataError("PARSE", "재무부 CSV에 Date 컬럼이 없습니다", "us-treasury")
for column, _key in TREASURY_COLUMNS:
if column not in index:
raise DataError(
"PARSE",
"재무부 CSV에 '{}' 컬럼이 없습니다".format(column),
"us-treasury",
)
records = []
for row in reader:
if not row:
continue
raw_date = row[index["Date"]] if index["Date"] < len(row) else ""
if not raw_date.strip():
continue
try:
iso_date = _parse_us_date(raw_date)
except ValueError:
continue
record = {"date": iso_date}
for column, key in TREASURY_COLUMNS:
position = index[column]
record[key] = _to_float(row[position]) if position < len(row) else None
records.append(record)
if not records:
raise DataError(
"EMPTY", "재무부 CSV에서 유효한 행을 찾지 못했습니다", "us-treasury"
)
records.sort(key=lambda item: item["date"])
return records
def extract_fred_payloads(raw):
"""FRED 응답에서 CSV 텍스트 목록을 꺼낸다.
FRED가 ZIP으로 응답하면 시리즈가 여러 CSV로 쪼개져 있으므로 전부 돌려준다.
하나만 고르면 나머지 시리즈가 조용히 사라진다.
"""
if not raw.startswith(b"PK\x03\x04"):
return [("<plain>", raw.decode("utf-8", errors="replace"))], False
try:
with zipfile.ZipFile(io.BytesIO(raw)) as archive:
payloads = []
for name in sorted(archive.namelist()):
if not name.lower().endswith(".csv"):
continue
payloads.append(
(name, archive.read(name).decode("utf-8", errors="replace"))
)
except zipfile.BadZipFile:
raise DataError("PARSE", "FRED ZIP 응답을 해석하지 못했습니다", "fred")
if not payloads:
raise DataError("PARSE", "FRED ZIP 응답에 CSV가 없습니다", "fred")
return payloads, True
def merge_fred_tables(tables):
"""FRED가 시리즈별 CSV로 쪼개 응답한 경우 날짜 기준으로 병합한다."""
ordered_ids = []
buckets = {}
for ids, rows in tables:
for series_id in ids:
if series_id not in ordered_ids:
ordered_ids.append(series_id)
for row in rows:
bucket = buckets.setdefault(row["date"], {})
for series_id, value in row["values"].items():
if value is not None or series_id not in bucket:
bucket[series_id] = value
merged = [
{
"date": day,
"values": {
series_id: buckets[day].get(series_id) for series_id in ordered_ids
},
}
for day in sorted(buckets)
]
return ordered_ids, merged
def parse_fred_csv(text):
"""FRED CSV 텍스트 -> (series ids, 날짜 오름차순 rows)."""
reader = csv.reader(io.StringIO(text))
try:
header = [column.strip() for column in next(reader)]
except StopIteration:
raise DataError("PARSE", "FRED CSV가 비어 있습니다", "fred")
if not header or header[0] != "observation_date" or len(header) < 2:
raise DataError(
"PARSE", "FRED CSV 헤더가 observation_date로 시작하지 않습니다", "fred"
)
ids = header[1:]
rows = []
for row in reader:
if not row or not row[0].strip():
continue
raw_date = row[0].strip()
if not re.match(r"^\d{4}-\d{2}-\d{2}$", raw_date):
continue
values = {}
for position, series_id in enumerate(ids):
cell = row[position + 1] if position + 1 < len(row) else ""
values[series_id] = _to_float(cell)
rows.append({"date": raw_date, "values": values})
if not rows:
raise DataError("EMPTY", "FRED CSV에서 유효한 행을 찾지 못했습니다", "fred")
rows.sort(key=lambda item: item["date"])
return ids, rows
def parse_cboe_csv(text):
"""Cboe VIX daily CSV -> 날짜/종가 레코드."""
reader = csv.DictReader(io.StringIO(text.lstrip("\ufeff")))
if not reader.fieldnames or not {"DATE", "CLOSE"}.issubset(
{name.strip().upper() for name in reader.fieldnames}
):
raise DataError("PARSE", "Cboe VIX CSV 필수 컬럼이 없습니다", "cboe-vix")
rows = []
for raw in reader:
normalized = {str(key).strip().upper(): value for key, value in raw.items()}
try:
observed = _parse_us_date(normalized.get("DATE", ""))
except ValueError:
continue
value = _to_float(normalized.get("CLOSE", ""))
if value is not None:
rows.append({"date": observed, "value": value})
if not rows:
raise DataError("EMPTY", "Cboe VIX CSV에 유효한 행이 없습니다", "cboe-vix")
return sorted(rows, key=lambda row: row["date"])
def parse_ecb_csv(text):
"""ECB EXR csvdata -> 날짜별 USD/JPY/GBP의 EUR 기준 환율."""
reader = csv.DictReader(io.StringIO(text.lstrip("\ufeff")))
required = {"TIME_PERIOD", "OBS_VALUE", "CURRENCY"}
if not reader.fieldnames or not required.issubset(set(reader.fieldnames)):
raise DataError("PARSE", "ECB CSV 필수 컬럼이 없습니다", "ecb-fx")
buckets = {}
for raw in reader:
observed = (raw.get("TIME_PERIOD") or "").strip()
currency = (raw.get("CURRENCY") or "").strip()
value = _to_float(raw.get("OBS_VALUE") or "")
if (
re.match(r"^\d{4}-\d{2}-\d{2}$", observed)
and currency
and value is not None
):
buckets.setdefault(observed, {})[currency] = value
rows = [{"date": day, "values": buckets[day]} for day in sorted(buckets)]
if not rows:
raise DataError("EMPTY", "ECB CSV에 유효한 행이 없습니다", "ecb-fx")
return rows
def parse_ecos_json(text):
"""ECOS JSON -> KOSPI/KOSDAQ 관측치. JSON 오류도 typed failure로 변환한다."""
try:
payload = json.loads(text)
except json.JSONDecodeError as error:
raise DataError("PARSE", "ECOS JSON 파싱 실패: {}".format(error.msg), "ecos")
if isinstance(payload, dict) and "RESULT" in payload:
result = payload["RESULT"]
raise DataError(
result.get("CODE", "ECOS"), result.get("MESSAGE", "ECOS 오류"), "ecos"
)
rows = []
for raw in payload.get("StatisticSearch", {}).get("row", []):
compact = str(raw.get("TIME") or "")
if not re.match(r"^\d{8}$", compact):
continue
label = str(raw.get("ITEM_NAME1") or raw.get("ITEM_CODE1") or "")
value = _to_float(str(raw.get("DATA_VALUE") or ""))
if value is not None and (
"KOSPI" in label.upper() or "KOSDAQ" in label.upper()
):
rows.append(
{
"date": "{}-{}-{}".format(compact[:4], compact[4:6], compact[6:]),
"symbol": "KOSDAQ" if "KOSDAQ" in label.upper() else "KOSPI",
"value": value,
}
)
if not rows:
raise DataError("EMPTY", "ECOS 응답에 KOSPI/KOSDAQ 관측치가 없습니다", "ecos")
return sorted(rows, key=lambda row: row["date"])
def latest_before(rows, cutoff):
candidates = [row for row in rows if row["date"] < cutoff]
return candidates[-1] if candidates else None
def ecb_crosses(row):
values = row["values"]
usd, jpy, gbp = values.get("USD"), values.get("JPY"), values.get("GBP")
result = {}
if usd is not None:
result["EUR/USD"] = {"value": round(usd, 6), "unit": "USD/EUR"}
if usd and jpy is not None:
result["USD/JPY"] = {"value": round(jpy / usd, 6), "unit": "JPY/USD"}
if gbp and usd is not None:
result["GBP/USD"] = {"value": round(usd / gbp, 6), "unit": "USD/GBP"}
return result
# --------------------------------------------------------------------------
# 계산
# --------------------------------------------------------------------------
def classify_curve(delta_2y_bp, delta_10y_bp):
"""2Y·10Y bp 변화로 curve를 판정한다."""
if delta_2y_bp is None or delta_10y_bp is None:
return None, None
if abs(delta_2y_bp) < CURVE_EPSILON_BP and abs(delta_10y_bp) < CURVE_EPSILON_BP:
return "flat", "Flat"
if abs(delta_2y_bp - delta_10y_bp) < CURVE_EPSILON_BP:
return "flat", "Flat"
if delta_2y_bp > 0 and delta_10y_bp > 0:
if delta_2y_bp > delta_10y_bp:
return "bear_flattening", "Bear Flattening"
return "bear_steepening", "Bear Steepening"
if delta_2y_bp < 0 and delta_10y_bp < 0:
if abs(delta_2y_bp) > abs(delta_10y_bp):
return "bull_steepening", "Bull Steepening"
return "bull_flattening", "Bull Flattening"
return "mixed", "Mixed"
def compute_yield_rows(records):
"""원자료 레코드에 spread·bp 변화·curve 판정을 붙인다."""
rows = []
previous = None
for record in records:
y2 = record.get("y2")
y10 = record.get("y10")
row = {
"date": record["date"],
"y2": _round(y2, 3),
"y10": _round(y10, 3),
"y30": _round(record.get("y30"), 3),
}
if y2 is not None and y10 is not None:
row["spread_2s10s_bp"] = _round((y10 - y2) * 100, 1)
else:
row["spread_2s10s_bp"] = None
previous_y2 = previous.get("y2") if previous else None
previous_y10 = previous.get("y10") if previous else None
if None not in (y2, y10, previous_y2, previous_y10):
delta_2y = (y2 - previous_y2) * 100
delta_10y = (y10 - previous_y10) * 100
curve, label = classify_curve(delta_2y, delta_10y)
row["d_y2_bp"] = _round(delta_2y, 1)
row["d_y10_bp"] = _round(delta_10y, 1)
row["d_spread_2s10s_bp"] = _round(delta_10y - delta_2y, 1)
row["curve"] = curve
row["curve_label"] = label
else:
row["d_y2_bp"] = None
row["d_y10_bp"] = None
row["d_spread_2s10s_bp"] = None
row["curve"] = None
row["curve_label"] = None
rows.append(row)
previous = record
return rows
def assess_staleness(latest_date, session):
"""관측일이 목표 세션보다 이전이면 (True, 지연일수)."""
if not latest_date or not session:
return False, 0
try:
lag = (_parse_iso(session) - _parse_iso(latest_date)).days
except ValueError:
return False, 0
if lag > 0:
return True, lag
return False, 0
def trim_rows(rows, start=None, end=None, last=None):
selected = rows
if start:
selected = [row for row in selected if row["date"] >= start]
if end:
selected = [row for row in selected if row["date"] <= end]
if last is not None and last > 0:
selected = selected[-last:]
return selected
def summarize_series(rows, series_id, session):
latest_date = None
latest_value = None
for row in rows:
value = row["values"].get(series_id)
if value is not None:
latest_date = row["date"]
latest_value = value
stale, lag_days = assess_staleness(latest_date, session)
return {
"id": series_id,
"latest_date": latest_date,
"latest_value": _round(latest_value, 4),
"stale": stale,
"lag_days": lag_days,
}
def _round(value, digits):
return None if value is None else round(value, digits)
def _now_utc():
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
# --------------------------------------------------------------------------
# 출력
# --------------------------------------------------------------------------
def render_yields_text(envelope):
lines = ["미 국채 par yield curve (출처: U.S. Department of the Treasury)", ""]
if envelope["session"]:
lines.append("목표 세션: {}".format(envelope["session"]))
lines.append("")
lines.append(
"날짜 2Y 10Y 30Y 2s10s(bp) Δ2Y Δ10Y Δ2s10s curve"
)
for row in envelope["rows"]:
lines.append(
"{date} {y2:>6} {y10:>6} {y30:>6} {spread:>9} {d2:>6} {d10:>6} {ds:>7} {curve}".format(
date=row["date"],
y2=_fmt(row["y2"], 3),
y10=_fmt(row["y10"], 3),
y30=_fmt(row["y30"], 3),
spread=_fmt(row["spread_2s10s_bp"], 1),
d2=_fmt(row["d_y2_bp"], 1),
d10=_fmt(row["d_y10_bp"], 1),
ds=_fmt(row["d_spread_2s10s_bp"], 1),
curve=row["curve_label"] or "-",
)
)
_append_warnings(lines, envelope["warnings"])
return "\n".join(lines)
def render_series_text(envelope):
lines = ["FRED 공개 시계열 (출처: Federal Reserve Bank of St. Louis)", ""]
if envelope["session"]:
lines.append("목표 세션: {}".format(envelope["session"]))
lines.append("")
lines.append("시리즈 최신 관측일 최신값 신선도")
for item in envelope["series"]:
freshness = (
"OK" if not item["stale"] else "STALE ({}일 지연)".format(item["lag_days"])
)
lines.append(
"{id:<12} {date:<12} {value:>12} {freshness}".format(
id=item["id"],
date=item["latest_date"] or "-",
value=_fmt(item["latest_value"], 4),
freshness=freshness,
)
)
lines.append("")
lines.append(
"날짜 "
+ " ".join("{:>12}".format(item["id"]) for item in envelope["series"])
)
for row in envelope["rows"]:
cells = " ".join(
"{:>12}".format(_fmt(row["values"].get(item["id"]), 4))
for item in envelope["series"]
)
lines.append("{} {}".format(row["date"], cells))
_append_warnings(lines, envelope["warnings"])
return "\n".join(lines)
def _fmt(value, digits):
return "-" if value is None else ("{:." + str(digits) + "f}").format(value)
def _append_warnings(lines, warnings):
if warnings:
lines.append("")
lines.append("경고:")
for warning in warnings:
lines.append("- {}".format(warning))
def emit(envelope, as_text, text_renderer):
if as_text:
print(text_renderer(envelope))
else:
print(json.dumps(envelope, ensure_ascii=False, indent=2))
# --------------------------------------------------------------------------
# 커맨드
# --------------------------------------------------------------------------
def build_yields_envelope(args):
year = args.year or date.today().year
url = TREASURY_URL.format(year=year)
raw = fetch_with_retry(url, "us-treasury", timeout=args.timeout)
records = parse_treasury_csv(raw.decode("utf-8", errors="replace"))
rows = compute_yield_rows(records)
warnings = []
if args.session:
latest = rows[-1]["date"]
stale, lag_days = assess_staleness(latest, args.session)
if stale:
warnings.append(
"최신 관측일 {}이(가) 목표 세션 {}보다 {}일 이전입니다. 이 값을 해당 세션 값으로 쓰지 마세요.".format(
latest, args.session, lag_days
)
)
trimmed = trim_rows(rows, start=args.start, end=args.end, last=args.last)
if not trimmed:
raise DataError(
"EMPTY", "지정한 기간에 해당하는 금리 관측치가 없습니다", "us-treasury"
)
return {
"ok": True,
"command": "yields",
"session": args.session,
"source": {"name": "us-treasury", "url": url, "fetched_at": _now_utc()},
"rows": trimmed,
"series": [],
"warnings": warnings,
}
def build_series_envelope(args):
ids = [item.strip() for item in args.ids.split(",") if item.strip()]
if not ids:
raise DataError(
"USAGE", "--ids에 시리즈 ID를 하나 이상 지정해야 합니다", "fred"
)
url = FRED_URL.format(ids=",".join(ids))
raw = fetch_with_retry(url, "fred", timeout=args.timeout)
payloads, was_zipped = extract_fred_payloads(raw)
parsed_ids, rows = merge_fred_tables(
[parse_fred_csv(text) for _name, text in payloads]
)
warnings = []
if was_zipped:
warnings.append(
"FRED 응답이 ZIP이어서 내부 CSV {}개를 병합했습니다.".format(len(payloads))
)
missing = [series_id for series_id in ids if series_id not in parsed_ids]
if missing:
warnings.append(
"요청한 시리즈 {}가 응답에 없습니다. 해당 시리즈는 본문에 쓰지 마세요.".format(
", ".join(missing)
)
)
if len(rows) > 200:
warnings.append(
"FRED CSV endpoint는 기간 파라미터를 무시하고 전체 이력을 반환합니다. helper가 {}개 행에서 클라이언트 절단을 수행했습니다.".format(
len(rows)
)
)
series = [
summarize_series(rows, series_id, args.session) for series_id in parsed_ids
]
for item in series:
if item["stale"]:
warnings.append(
"{} 최신 관측일 {}이(가) 목표 세션 {}보다 {}일 이전입니다. 이 값을 해당 세션 값으로 쓰지 마세요.".format(
item["id"], item["latest_date"], args.session, item["lag_days"]
)
)
trimmed = trim_rows(rows, start=args.start, end=args.end, last=args.last)
if not trimmed:
raise DataError(
"EMPTY", "지정한 기간에 해당하는 FRED 관측치가 없습니다", "fred"
)
return {
"ok": True,
"command": "series",
"session": args.session,
"source": {"name": "fred", "url": url, "fetched_at": _now_utc()},
"rows": trimmed,
"series": series,
"warnings": warnings,
}
def _dated_source_envelope(command, source, url, rows, session, last):
selected = trim_rows(rows, last=last)
latest = rows[-1]
stale, lag_days = assess_staleness(latest["date"], session)
warnings = []
if stale:
warnings.append(
"최신 관측일 {}이(가) 목표 세션 {}보다 {}일 이전입니다.".format(
latest["date"], session, lag_days
)
)
return {
"ok": True,
"command": command,
"session": session,
"source": {"name": source, "url": url, "fetched_at": _now_utc()},
"rows": selected,
"stale": stale,
"lag_days": lag_days,
"warnings": warnings,
}
def build_cboe_envelope(args):
raw = fetch_with_retry(CBOE_VIX_URL, "cboe-vix", timeout=args.timeout)
rows = parse_cboe_csv(raw.decode("utf-8", errors="replace"))
envelope = _dated_source_envelope(
"cboe-vix", "Cboe VIX", CBOE_VIX_URL, rows, args.session, args.last
)
for row in envelope["rows"]:
row.update(symbol="VIX", unit="pt", basis="close")
return envelope
def build_ecb_envelope(args):
end = args.end or date.today().isoformat()
start = args.start or (date.fromisoformat(end) - timedelta(days=14)).isoformat()
url = ECB_FX_URL.format(start=start, end=end)
raw = fetch_with_retry(url, "ecb-fx", timeout=args.timeout)
rows = parse_ecb_csv(raw.decode("utf-8", errors="replace"))
converted = [
{"date": row["date"], "values": ecb_crosses(row), "basis": "reference-rate"}
for row in rows
]
return _dated_source_envelope(
"ecb-fx", "ECB reference rates", url, converted, args.session, args.last
)
def _failure(error, section):
return {
"source": error.source,
"error": error.code,
"message": error.message,
"section": section,
}
def build_snapshot_envelope(args):
"""공식 원천을 독립 호출해 한 원천 실패가 다른 자산군을 중단하지 않게 한다."""
briefing = date.fromisoformat(args.briefing_date)
cutoff = briefing.isoformat()
start = (briefing - timedelta(days=21)).isoformat()
items, warnings, failures = [], [], []
sessions = {"us_overnight": None, "korea_previous": None}
try:
raw = fetch_with_retry(CBOE_VIX_URL, "cboe-vix", timeout=args.timeout)
row = latest_before(
parse_cboe_csv(raw.decode("utf-8", errors="replace")), cutoff
)
if not row:
raise DataError("EMPTY", "브리핑일 이전 Cboe 관측치가 없습니다", "cboe-vix")
sessions["us_overnight"] = row["date"]
items.append(
{
"symbol": "VIX",
"section": "Equity, Vol",
"value": row["value"],
"unit": "pt",
"observed": row["date"],
"session": row["date"],
"basis": "close",
"source_name": "Cboe",
"source_url": CBOE_VIX_URL,
}
)
except DataError as error:
failures.append(_failure(error, "Equity, Vol"))
ecos_url = None
try:
rows, row = [], None
candidate = briefing - timedelta(days=1)
for _ in range(14):
if candidate.weekday() < 5:
compact = candidate.strftime("%Y%m%d")
ecos_url = ECOS_URL.format(start=compact, end=compact)
try:
raw = fetch_with_retry(
ecos_url, "ecos", timeout=args.timeout, attempts=1
)
rows = parse_ecos_json(raw.decode("utf-8", errors="replace"))
kospi_rows = [
market_row
for market_row in rows
if market_row["symbol"] == "KOSPI"
]
row = kospi_rows[-1] if kospi_rows else None
except DataError as candidate_error:
if candidate_error.code not in {"EMPTY", "INFO-200"}:
raise
if row:
break
candidate -= timedelta(days=1)
if not row:
raise DataError("EMPTY", "브리핑일 이전 KOSPI 관측치가 없습니다", "ecos")
sessions["korea_previous"] = row["date"]
for market_row in rows:
if market_row["date"] == row["date"]:
items.append(
{
"symbol": market_row["symbol"],
"section": "한국 증시",
"value": market_row["value"],
"unit": "pt",
"observed": market_row["date"],
"session": row["date"],
"basis": "close",
"source_name": "한국은행 ECOS",
"source_url": ecos_url,
}
)
except DataError as error:
failures.append(_failure(error, "한국 증시"))
us_session = sessions["us_overnight"]
if us_session:
try:
url = TREASURY_URL.format(year=briefing.year)
raw = fetch_with_retry(url, "us-treasury", timeout=args.timeout)
records = [
row
for row in parse_treasury_csv(raw.decode("utf-8", errors="replace"))
if row["date"] <= us_session
]
matched = (
records[-1] if records and records[-1]["date"] == us_session else None
)
if not matched:
warnings.append(
"Rates: 미국 세션과 일치하는 Treasury 관측치가 없어 제외했습니다."
)
else:
computed = compute_yield_rows(records)
latest = computed[-1]
for symbol, field, unit in (
("2Y", "y2", "%"),
("10Y", "y10", "%"),
("30Y", "y30", "%"),
("2s10s", "spread_2s10s_bp", "bp"),
):
if latest[field] is not None:
items.append(
{
"symbol": symbol,
"section": "Rates",
"value": latest[field],
"unit": unit,
"observed": latest["date"],
"session": us_session,
"basis": "close",
"source_name": "U.S. Treasury",
"source_url": url,
}
)
except DataError as error:
failures.append(_failure(error, "Rates"))
try:
fx_url = ECB_FX_URL.format(start=start, end=cutoff)
raw = fetch_with_retry(fx_url, "ecb-fx", timeout=args.timeout)
fx_rows = parse_ecb_csv(raw.decode("utf-8", errors="replace"))
fx_row = next(
(row for row in reversed(fx_rows) if row["date"] <= us_session), None
)
if not fx_row or fx_row["date"] != us_session:
warnings.append(
"FX: 미국 세션과 일치하는 ECB reference rate가 없어 제외했습니다."
)
else:
for symbol, quote in ecb_crosses(fx_row).items():
items.append(
{
"symbol": symbol,
"section": "FX",
"value": quote["value"],
"unit": quote["unit"],
"observed": fx_row["date"],
"session": us_session,
"basis": "reference-rate",
"source_name": "ECB",
"source_url": fx_url,
}
)
except DataError as error:
failures.append(_failure(error, "FX"))
return {
"version": 1,
"command": "snapshot",
"generated_at": _now_utc(),
"briefing_date": cutoff,
"sessions": sessions,
"items": items,
"warnings": warnings,
"failures": failures,
}
# --------------------------------------------------------------------------
# CLI
# --------------------------------------------------------------------------
def build_parser():
parser = argparse.ArgumentParser(
prog="market_data.py",
description="멀티에셋 모닝 브리핑용 공개 시장 데이터 helper",
)
subparsers = parser.add_subparsers(dest="command")
yields_parser = subparsers.add_parser(
"yields", help="미국 재무부 일별 par yield curve"
)
yields_parser.add_argument(
"--year", type=int, default=None, help="조회 연도 (기본: 올해)"
)
_add_common_arguments(yields_parser)
series_parser = subparsers.add_parser("series", help="FRED 공개 시계열")
series_parser.add_argument(
"--ids", required=True, help="쉼표로 구분한 FRED 시리즈 ID"
)
_add_common_arguments(series_parser)
cboe_parser = subparsers.add_parser("cboe-vix", help="Cboe 공식 VIX 일별 종가")
_add_common_arguments(cboe_parser)
ecb_parser = subparsers.add_parser("ecb-fx", help="ECB 공식 일별 reference rate")
_add_common_arguments(ecb_parser)
snapshot_parser = subparsers.add_parser(
"snapshot", help="독립 세션과 공식 기초 데이터"
)
snapshot_parser.add_argument(
"--briefing-date", required=True, help="서울 기준일 YYYY-MM-DD"
)
snapshot_parser.add_argument(
"--timeout", type=float, default=DEFAULT_TIMEOUT, help="HTTP 타임아웃(초)"
)
return parser
def _add_common_arguments(parser):
parser.add_argument("--last", type=int, default=None, help="마지막 N개 관측치만")
parser.add_argument("--start", default=None, help="시작일 YYYY-MM-DD")
parser.add_argument("--end", default=None, help="종료일 YYYY-MM-DD")
parser.add_argument("--session", default=None, help="목표 세션 날짜 YYYY-MM-DD")
parser.add_argument(
"--timeout", type=float, default=DEFAULT_TIMEOUT, help="HTTP 타임아웃(초)"
)
parser.add_argument("--text", action="store_true", help="JSON 대신 사람이 읽는 표")
def main(argv=None):
parser = build_parser()
args = parser.parse_args(argv)
if not args.command:
parser.print_help()
return 2
try:
if args.command == "yields":
envelope = build_yields_envelope(args)
emit(envelope, args.text, render_yields_text)
elif args.command == "series":
envelope = build_series_envelope(args)
emit(envelope, args.text, render_series_text)
elif args.command == "cboe-vix":
print(json.dumps(build_cboe_envelope(args), ensure_ascii=False, indent=2))
elif args.command == "ecb-fx":
print(json.dumps(build_ecb_envelope(args), ensure_ascii=False, indent=2))
else:
print(
json.dumps(build_snapshot_envelope(args), ensure_ascii=False, indent=2)
)
except ValueError as error:
data_error = DataError("USAGE", str(error), "cli")
print(
json.dumps(
{
"ok": False,
"error": {
"code": data_error.code,
"message": data_error.message,
"source": data_error.source,
},
},
ensure_ascii=False,
indent=2,
),
file=sys.stderr,
)
return 1
except DataError as error:
if getattr(args, "text", False):
print(
"오류[{}] {}: {}".format(error.source, error.code, error.message),
file=sys.stderr,
)
else:
print(
json.dumps(
{
"ok": False,
"error": {
"code": error.code,
"message": error.message,
"source": error.source,
},
},
ensure_ascii=False,
indent=2,
),
file=sys.stderr,
)
return 1
return 0
if __name__ == "__main__":
sys.exit(main())
skill.json
{
"name": "multi-asset-morning-briefing",
"description": "글로벌 주식·금리·외환·원자재·변동성과 한국시장 전달 경로를 연결해 근거 링크가 있는 7개 섹션의 멀티에셋 모닝 브리핑을 작성한다. Use when the user asks 모닝 브리핑, Morning Call, 오버나이트 시황, 데일리 마켓 브리핑 또는 기존 브리핑 검수. Not for 실시간 호가, 국내 개별 종목 단독 조회(korean-stock-search), 투자 권유·매매 판단.",
"profiles": [
"lookup",
"browser"
],
"frontmatter": "name: multi-asset-morning-briefing\ndescription: 글로벌 주식·금리·외환·원자재·변동성과 한국시장 전달 경로를 연결해 근거 링크가 있는 7개 섹션의 멀티에셋 모닝 브리핑을 작성한다. Use when the user asks 모닝 브리핑, Morning Call, 오버나이트 시황, 데일리 마켓 브리핑 또는 기존 브리핑 검수. Not for 실시간 호가, 국내 개별 종목 단독 조회(korean-stock-search), 투자 권유·매매 판단.\nlicense: MIT\nmetadata:\n category: finance\n locale: ko-KR\n phase: v1"
}
SKILL.md
---
name: multi-asset-morning-briefing
description: 글로벌 주식·금리·외환·원자재·변동성과 한국시장 전달 경로를 연결해 근거 링크가 있는 7개 섹션의 멀티에셋 모닝 브리핑을 작성한다. Use when the user asks 모닝 브리핑, Morning Call, 오버나이트 시황, 데일리 마켓 브리핑 또는 기존 브리핑 검수. Not for 실시간 호가, 국내 개별 종목 단독 조회(korean-stock-search), 투자 권유·매매 판단.
license: MIT
metadata:
category: finance
locale: ko-KR
phase: v1
---
# multi-asset-morning-briefing
<!-- k-skill:cli-stub — generated by scripts/generate-skill-stubs.js; edit skill.json / instruction.md instead -->
## Get the full instructions (required first step)
Run this and follow its output as the primary instructions for this skill:
```bash
npx -y @nomadamas/k-skill@0 instruct multi-asset-morning-briefing
```
The CLI detects the current runtime (Dolshoi vault/CloakBrowser vs generic) and prints only the applicable instructions, always up to date. Helper files bundled with the CLI are listed by:
```bash
npx -y @nomadamas/k-skill@0 files multi-asset-morning-briefing
```
Keep the CLI and every coding-agent skill install current (including Vercel Agent Skills) with:
```bash
npx -y @nomadamas/k-skill@0 update
```
If `npx` is unavailable, install Node.js 18+ or follow https://github.com/NomaDamas/k-skill#readme, or read the source instructions at https://github.com/NomaDamas/k-skill/blob/main/multi-asset-morning-briefing/instruction.md.
## Hard rules even without the CLI
- Never execute payment, message/email delivery, final submission, cancellation, or public posting without the user's explicit approval immediately beforehand.
- Never ask for, print, or store plaintext credentials in chat, files, or shell arguments.
- Never bypass legal, physical-presence, CAPTCHA, identity-proofing, or electronic-signature boundaries.
tests/fixtures/sample_briefing.md
2026.09.17 Morning Market Briefing
1. Summary
• 미국 주식 약세와 VIX 상승이 위험선호를 제약해 한국 성장주에 부담 [Cboe](https://www.cboe.com/tradable_products/vix/)
2. Rates
• 미 국채 2년·10년 금리 상승과 2s10s 확대가 장기물 약세를 시사하는 핵심 [U.S. Treasury](https://home.treasury.gov/resource-center/data-chart-center/interest-rates)
3. FX
• EUR/USD 하락과 USD/JPY 상승이 달러 강세 압력으로 연결될 가능성 [ECB](https://data.ecb.europa.eu/data/datasets/EXR)
4. Commodity
• WTI·Brent 상승이 에너지 업종에는 긍정적이나 운송비에는 부담 [EIA](https://www.eia.gov/petroleum/)
5. Equity, Vol
• S&P 500 하락과 VIX 상승 조합으로 방어주 상대강도 확대 [Cboe](https://www.cboe.com/tradable_products/vix/)
6. 한국 증시
• 원/달러 상승과 미국 반도체 약세가 외국인 수급과 대형 기술주에 부담 [한국은행](https://ecos.bok.or.kr/)
7. 주요 일정
• 2026.09.17 21:30 KST 미국 고용지표 발표가 금리와 달러의 당일 핵심 변수 [BLS](https://www.bls.gov/schedule/)
tests/test_market_data.py
#!/usr/bin/env python3
"""market_data.py 오프라인 단위 테스트. 네트워크를 타지 않는다."""
import contextlib
import io
import json
import os
import re
import sys
import unittest
import urllib.error
import zipfile
from pathlib import Path
from unittest.mock import patch
sys.path.insert(
0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "scripts")
)
import market_data # noqa: E402
TREASURY_FIXTURE = (
'Date,"1 Mo","2 Mo","3 Mo","6 Mo","1 Yr","2 Yr","3 Yr","5 Yr","7 Yr","10 Yr","20 Yr","30 Yr"\n'
"09/15/2026,3.93,4.06,4.11,4.17,4.39,4.67,4.76,4.83,4.91,5.00,5.40,5.36\n"
"09/14/2026,3.94,4.06,4.11,4.18,4.37,4.65,4.73,4.80,4.88,4.97,5.37,5.34\n"
"09/11/2026,3.93,4.05,4.07,4.12,4.35,4.63,4.69,4.78,4.87,4.96,5.38,5.35\n"
)
FRED_FIXTURE = (
"observation_date,SP500,VIXCLS,DEXKOUS\n"
"2026-09-11,7656.98,15.10,1340.30\n"
"2026-09-14,7619.98,,1385.09\n"
"2026-09-15,7585.73,17.20,.\n"
)
CBOE_FIXTURE = (
"DATE,OPEN,HIGH,LOW,CLOSE\n09/10/2026,15,17,14,16.2\n09/11/2026,16,18,15,17.1\n"
)
ECB_FIXTURE = (
"TIME_PERIOD,OBS_VALUE,CURRENCY\n"
"2026-09-11,1.18,USD\n2026-09-11,173.46,JPY\n2026-09-11,0.86,GBP\n"
)
ECOS_FIXTURE = json.dumps(
{
"StatisticSearch": {
"row": [
{"TIME": "20260910", "ITEM_NAME1": "KOSPI지수", "DATA_VALUE": "3300.1"},
{"TIME": "20260911", "ITEM_NAME1": "KOSPI지수", "DATA_VALUE": "3310.2"},
{"TIME": "20260911", "ITEM_NAME1": "KOSDAQ지수", "DATA_VALUE": "910.3"},
]
}
}
)
class TreasuryParseTests(unittest.TestCase):
def test_parses_and_sorts_ascending(self):
records = market_data.parse_treasury_csv(TREASURY_FIXTURE)
self.assertEqual(
[item["date"] for item in records],
["2026-09-11", "2026-09-14", "2026-09-15"],
)
self.assertEqual(records[-1]["y2"], 4.67)
self.assertEqual(records[-1]["y10"], 5.00)
self.assertEqual(records[-1]["y30"], 5.36)
def test_missing_column_is_parse_error(self):
with self.assertRaises(market_data.DataError) as context:
market_data.parse_treasury_csv('Date,"2 Yr"\n09/15/2026,4.67\n')
self.assertEqual(context.exception.code, "PARSE")
self.assertEqual(context.exception.source, "us-treasury")
def test_empty_input_is_parse_error(self):
with self.assertRaises(market_data.DataError) as context:
market_data.parse_treasury_csv("")
self.assertEqual(context.exception.code, "PARSE")
class FredParseTests(unittest.TestCase):
def test_blank_and_dot_values_become_none(self):
ids, rows = market_data.parse_fred_csv(FRED_FIXTURE)
self.assertEqual(ids, ["SP500", "VIXCLS", "DEXKOUS"])
self.assertEqual(rows[-1]["values"]["VIXCLS"], 17.20)
self.assertIsNone(rows[-1]["values"]["DEXKOUS"])
self.assertIsNone(rows[-2]["values"]["VIXCLS"])
def test_bad_header_is_parse_error(self):
with self.assertRaises(market_data.DataError) as context:
market_data.parse_fred_csv("date,VALUE\n2026-09-15,1.0\n")
self.assertEqual(context.exception.code, "PARSE")
self.assertEqual(context.exception.source, "fred")
def test_plain_payload_is_single_table(self):
payloads, was_zipped = market_data.extract_fred_payloads(
FRED_FIXTURE.encode("utf-8")
)
self.assertFalse(was_zipped)
self.assertEqual(len(payloads), 1)
ids, rows = market_data.parse_fred_csv(payloads[0][1])
self.assertEqual(ids, ["SP500", "VIXCLS", "DEXKOUS"])
self.assertEqual(len(rows), 3)
def test_split_zip_members_are_merged_without_losing_series(self):
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w") as archive:
archive.writestr("README.txt", "source: FRED")
archive.writestr(
"sp500.csv",
"observation_date,SP500\n2026-09-11,7656.98\n2026-09-15,7585.73\n",
)
archive.writestr(
"vix.csv",
"observation_date,VIXCLS\n2026-09-11,15.10\n2026-09-14,16.40\n",
)
payloads, was_zipped = market_data.extract_fred_payloads(buffer.getvalue())
self.assertTrue(was_zipped)
self.assertEqual(len(payloads), 2)
ids, rows = market_data.merge_fred_tables(
[market_data.parse_fred_csv(text) for _name, text in payloads]
)
self.assertEqual(ids, ["SP500", "VIXCLS"])
self.assertEqual(
[row["date"] for row in rows], ["2026-09-11", "2026-09-14", "2026-09-15"]
)
self.assertEqual(rows[1]["values"], {"SP500": None, "VIXCLS": 16.40})
self.assertEqual(rows[-1]["values"], {"SP500": 7585.73, "VIXCLS": None})
class OfficialSourceParseTests(unittest.TestCase):
def test_cboe_vix_and_asset_unit(self):
rows = market_data.parse_cboe_csv(CBOE_FIXTURE)
self.assertEqual(rows[-1], {"date": "2026-09-11", "value": 17.1})
def test_ecb_cross_rates_have_pair_units(self):
rows = market_data.parse_ecb_csv(ECB_FIXTURE)
crosses = market_data.ecb_crosses(rows[-1])
self.assertEqual(crosses["EUR/USD"]["unit"], "USD/EUR")
self.assertEqual(crosses["USD/JPY"]["unit"], "JPY/USD")
self.assertAlmostEqual(crosses["USD/JPY"]["value"], 147.0)
def test_malformed_ecos_json_is_typed(self):
with self.assertRaises(market_data.DataError) as context:
market_data.parse_ecos_json("{not-json")
self.assertEqual(
(context.exception.code, context.exception.source), ("PARSE", "ecos")
)
class IndependentSessionTests(unittest.TestCase):
def test_snapshot_resolves_us_and_korea_sessions_independently(self):
cboe_us_holiday = "DATE,OPEN,HIGH,LOW,CLOSE\n09/10/2026,15,17,14,16.2\n"
def fake_fetch(url, timeout=market_data.DEFAULT_TIMEOUT):
if "VIX_History" in url:
return cboe_us_holiday.encode()
if "ecos.bok" in url:
return ECOS_FIXTURE.encode()
if "treasury.gov" in url:
return TREASURY_FIXTURE.encode()
if "data-api.ecb" in url:
return ECB_FIXTURE.encode()
raise AssertionError(url)
original = market_data.fetch_bytes
market_data.fetch_bytes = fake_fetch
try:
args = type("Args", (), {"briefing_date": "2026-09-14", "timeout": 1})()
envelope = market_data.build_snapshot_envelope(args)
finally:
market_data.fetch_bytes = original
self.assertEqual(
envelope["sessions"],
{"us_overnight": "2026-09-10", "korea_previous": "2026-09-11"},
)
def test_one_source_failure_does_not_abort_snapshot(self):
def fake_fetch(url, timeout=market_data.DEFAULT_TIMEOUT):
if "VIX_History" in url:
raise urllib.error.URLError("offline")
if "ecos.bok" in url:
return ECOS_FIXTURE.encode()
raise urllib.error.URLError("offline")
original = market_data.fetch_bytes
market_data.fetch_bytes = fake_fetch
try:
args = type("Args", (), {"briefing_date": "2026-09-14", "timeout": 1})()
envelope = market_data.build_snapshot_envelope(args)
finally:
market_data.fetch_bytes = original
self.assertIsNone(envelope["sessions"]["us_overnight"])
self.assertEqual(envelope["sessions"]["korea_previous"], "2026-09-11")
self.assertTrue(envelope["failures"])
class CurveTests(unittest.TestCase):
def test_classification(self):
cases = [
(4.0, 3.0, "bear_flattening", "Bear Flattening"),
(3.0, 4.0, "bear_steepening", "Bear Steepening"),
(-4.0, -3.0, "bull_steepening", "Bull Steepening"),
(-3.0, -4.0, "bull_flattening", "Bull Flattening"),
(3.0, -2.0, "mixed", "Mixed"),
(0.0, 0.0, "flat", "Flat"),
(2.0, 2.0, "flat", "Flat"),
]
for delta_2y, delta_10y, expected_key, expected_label in cases:
with self.subTest(delta_2y=delta_2y, delta_10y=delta_10y):
self.assertEqual(
market_data.classify_curve(delta_2y, delta_10y),
(expected_key, expected_label),
)
def test_none_input_returns_none(self):
self.assertEqual(market_data.classify_curve(None, 1.0), (None, None))
self.assertEqual(market_data.classify_curve(1.0, None), (None, None))
class YieldRowTests(unittest.TestCase):
def test_spread_and_changes_use_raw_values(self):
records = [
{"date": "2026-09-14", "y2": 4.651, "y10": 4.985, "y30": 5.34},
{"date": "2026-09-15", "y2": 4.655, "y10": 4.990, "y30": 5.36},
]
rows = market_data.compute_yield_rows(records)
self.assertIsNone(rows[0]["d_y2_bp"])
self.assertIsNone(rows[0]["curve"])
self.assertEqual(rows[0]["spread_2s10s_bp"], 33.4)
self.assertEqual(rows[1]["d_y2_bp"], 0.4)
self.assertEqual(rows[1]["d_y10_bp"], 0.5)
self.assertEqual(rows[1]["d_spread_2s10s_bp"], 0.1)
self.assertEqual(rows[1]["spread_2s10s_bp"], 33.5)
self.assertEqual(rows[1]["curve"], "bear_steepening")
def test_missing_yield_does_not_break_spread(self):
records = [
{"date": "2026-09-14", "y2": None, "y10": 4.985, "y30": None},
{"date": "2026-09-15", "y2": 4.655, "y10": None, "y30": None},
]
rows = market_data.compute_yield_rows(records)
self.assertIsNone(rows[0]["spread_2s10s_bp"])
self.assertIsNone(rows[1]["spread_2s10s_bp"])
self.assertIsNone(rows[1]["d_spread_2s10s_bp"])
class StalenessTests(unittest.TestCase):
def test_lag_detection(self):
self.assertEqual(
market_data.assess_staleness("2026-09-11", "2026-09-15"), (True, 4)
)
self.assertEqual(
market_data.assess_staleness("2026-09-15", "2026-09-15"), (False, 0)
)
self.assertEqual(
market_data.assess_staleness("2026-09-16", "2026-09-15"), (False, 0)
)
self.assertEqual(market_data.assess_staleness(None, "2026-09-15"), (False, 0))
self.assertEqual(market_data.assess_staleness("2026-09-11", None), (False, 0))
def test_summary_skips_trailing_missing_values(self):
_, rows = market_data.parse_fred_csv(FRED_FIXTURE)
summary = market_data.summarize_series(rows, "DEXKOUS", "2026-09-15")
self.assertEqual(summary["latest_date"], "2026-09-14")
self.assertEqual(summary["latest_value"], 1385.09)
self.assertEqual((summary["stale"], summary["lag_days"]), (True, 1))
def test_summary_is_fresh_when_session_matches(self):
_, rows = market_data.parse_fred_csv(FRED_FIXTURE)
summary = market_data.summarize_series(rows, "SP500", "2026-09-15")
self.assertEqual((summary["stale"], summary["lag_days"]), (False, 0))
class TrimTests(unittest.TestCase):
def setUp(self):
self.rows = [{"date": "2026-09-{:02d}".format(day)} for day in (10, 11, 14, 15)]
def test_last(self):
self.assertEqual(
[row["date"] for row in market_data.trim_rows(self.rows, last=2)],
["2026-09-14", "2026-09-15"],
)
def test_start_and_end(self):
self.assertEqual(
[
row["date"]
for row in market_data.trim_rows(
self.rows, start="2026-09-11", end="2026-09-14"
)
],
["2026-09-11", "2026-09-14"],
)
class CliTests(unittest.TestCase):
def test_http_failure_returns_error_envelope_and_exit_code(self):
def boom(url, timeout=market_data.DEFAULT_TIMEOUT):
raise urllib.error.URLError("connection refused")
original = market_data.fetch_bytes
market_data.fetch_bytes = boom
stdout, stderr = io.StringIO(), io.StringIO()
try:
with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
exit_code = market_data.main(["series", "--ids", "SP500"])
finally:
market_data.fetch_bytes = original
self.assertEqual(exit_code, 1)
envelope = json.loads(stderr.getvalue())
self.assertFalse(envelope["ok"])
self.assertEqual(envelope["error"]["source"], "fred")
def test_yields_envelope_never_overwrites_observation_dates(self):
payload = TREASURY_FIXTURE.encode("utf-8")
original = market_data.fetch_bytes
market_data.fetch_bytes = lambda url, timeout=market_data.DEFAULT_TIMEOUT: (
payload
)
stdout = io.StringIO()
try:
with contextlib.redirect_stdout(stdout):
exit_code = market_data.main(
["yields", "--last", "2", "--session", "2026-09-16"]
)
finally:
market_data.fetch_bytes = original
self.assertEqual(exit_code, 0)
envelope = json.loads(stdout.getvalue())
self.assertEqual(
[row["date"] for row in envelope["rows"]], ["2026-09-14", "2026-09-15"]
)
self.assertTrue(envelope["warnings"])
self.assertEqual(envelope["rows"][1]["curve_label"], "Bear Steepening")
self.assertEqual(envelope["rows"][1]["d_spread_2s10s_bp"], 1.0)
def test_missing_requested_series_is_reported(self):
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w") as archive:
archive.writestr(
"sp500.csv", "observation_date,SP500\n2026-09-15,7585.73\n"
)
payload = buffer.getvalue()
original = market_data.fetch_bytes
market_data.fetch_bytes = lambda url, timeout=market_data.DEFAULT_TIMEOUT: (
payload
)
stdout = io.StringIO()
try:
with contextlib.redirect_stdout(stdout):
exit_code = market_data.main(
["series", "--ids", "SP500,VIXCLS", "--last", "1"]
)
finally:
market_data.fetch_bytes = original
self.assertEqual(exit_code, 0)
envelope = json.loads(stdout.getvalue())
self.assertEqual([item["id"] for item in envelope["series"]], ["SP500"])
self.assertTrue(any("VIXCLS" in warning for warning in envelope["warnings"]))
def test_timeout_is_typed_and_actionable(self):
def slow(url, timeout=market_data.DEFAULT_TIMEOUT):
raise TimeoutError("simulated timeout")
original = market_data.fetch_bytes
market_data.fetch_bytes = slow
stdout, stderr = io.StringIO(), io.StringIO()
try:
with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
exit_code = market_data.main(["yields"])
finally:
market_data.fetch_bytes = original
self.assertEqual(exit_code, 1)
envelope = json.loads(stderr.getvalue())
self.assertEqual(envelope["error"]["code"], "TIMEOUT")
self.assertIn("--timeout", envelope["error"]["message"])
def test_no_subcommand_is_usage_error(self):
stdout = io.StringIO()
with contextlib.redirect_stdout(stdout):
self.assertEqual(market_data.main([]), 2)
class NoNetworkTests(unittest.TestCase):
def test_pure_functions_do_not_touch_network(self):
def forbidden(*args, **kwargs):
raise AssertionError("pure functions must not open the network")
original = market_data.urllib.request.urlopen
market_data.urllib.request.urlopen = forbidden
try:
records = market_data.parse_treasury_csv(TREASURY_FIXTURE)
market_data.compute_yield_rows(records)
market_data.parse_fred_csv(FRED_FIXTURE)
market_data.assess_staleness("2026-09-11", "2026-09-15")
finally:
market_data.urllib.request.urlopen = original
class FakeResponse:
def __init__(self, body, declared=None):
self.body = body
self.headers = {} if declared is None else {"Content-Length": str(declared)}
self.read_sizes = []
def __enter__(self):
return self
def __exit__(self, *_args):
return None
def read(self, size=-1):
self.read_sizes.append(size)
chunk, self.body = self.body[:size], self.body[size:]
return chunk
class HttpSafetyTests(unittest.TestCase):
def test_declared_oversized_response_is_rejected_before_read(self):
response = FakeResponse(b"x", market_data.MAX_RESPONSE_BYTES + 1)
with patch("market_data.urllib.request.urlopen", return_value=response):
with self.assertRaises(market_data.DataError) as context:
market_data.fetch_bytes("https://example.com")
self.assertEqual(context.exception.code, "LIMIT")
self.assertEqual(response.read_sizes, [])
def test_streamed_oversized_response_uses_bounded_reads(self):
response = FakeResponse(b"x" * (market_data.MAX_RESPONSE_BYTES + 1))
with patch("market_data.urllib.request.urlopen", return_value=response):
with self.assertRaises(market_data.DataError) as context:
market_data.fetch_bytes("https://example.com")
self.assertEqual(context.exception.code, "LIMIT")
self.assertTrue(all(0 < size <= 64 * 1024 for size in response.read_sizes))
class BriefingContractTests(unittest.TestCase):
def test_golden_briefing_has_title_sections_links_and_memo_tone(self):
fixture = Path(__file__).with_name("fixtures") / "sample_briefing.md"
text = fixture.read_text(encoding="utf-8")
lines = text.splitlines()
self.assertRegex(lines[0], r"^\d{4}\.\d{2}\.\d{2} Morning Market Briefing$")
headings = [line for line in lines if re.match(r"^\d+\. ", line)]
self.assertEqual(
headings,
[
"1. Summary",
"2. Rates",
"3. FX",
"4. Commodity",
"5. Equity, Vol",
"6. 한국 증시",
"7. 주요 일정",
],
)
bullets = [line for line in lines if line.startswith("•")]
self.assertEqual(len(bullets), 7)
self.assertTrue(
all(re.search(r"\[[^]]+\]\(https://[^)]+\)$", line) for line in bullets)
)
self.assertFalse(re.search(r"(?i)(^|\s)by(?:\s|$)", text))
self.assertTrue(
all(
not re.search(r"(입니다|했습니다|이다|했다|합니다|한다)\.?\s*\[", line)
for line in bullets
)
)
if __name__ == "__main__":
unittest.main()