# -*- coding: utf-8 -*- """LICH_TIN.PY — lịch tin kinh tế cho bản tin, nguồn LỊCH MT5 (EA FeederPhu -> api1, có số thực tế + lịch sử). 02/10/2026 (bản 2): bỏ ForexFactory (chỉ có tuần hiện tại, không lịch sử). Lịch MT5 có số thực tế / dự báo / kỳ trước của từng tin từ 12/2022, và api1 tính sẵn PHẢN ỨNG CỦA VÀNG ở các lần ra tin trước (2 nến H1 từ giờ ra tin) -> bản tin nói được "tin này các lần trước làm vàng chạy bao nhiêu", và "số ra cao hơn dự báo thì vàng thường đi hướng nào". Chỉ lấy tin Mỹ ảnh hưởng tới vàng (TIN_VANG); nhiều tin cùng giờ gộp thành một mốc, phản ứng lấy theo tin chính. """ from __future__ import annotations import statistics from datetime import date, datetime, timedelta from typing import Any, Dict, List, Optional # mã sự kiện MT5 -> (tên tiếng Việt, hạng: 1 = tin rất mạnh với vàng, 2 = tin mạnh) TIN_VANG = { 'nonfarm-payrolls': ('Bảng lương phi nông nghiệp (NFP)', 1), 'unemployment-rate': ('Tỷ lệ thất nghiệp', 1), 'average-hourly-earnings-mm': ('Thu nhập bình quân theo giờ', 2), 'consumer-price-index-mm': ('Lạm phát CPI (tháng)', 1), 'consumer-price-index-yy': ('Lạm phát CPI (năm)', 1), 'consumer-price-index-ex-food-energy-mm': ('CPI lõi (tháng)', 1), 'core-pce-price-index-mm': ('PCE lõi (tháng)', 1), 'core-pce-price-index-yy': ('PCE lõi (năm)', 2), 'fed-interest-rate-decision': ('Quyết định lãi suất Fed', 1), 'fomc-meeting-statement': ('Tuyên bố FOMC', 1), 'fomc-press-conference': ('Họp báo FOMC', 1), 'fomc-economic-projections': ('Dự phóng kinh tế FOMC', 1), 'fomc-minutes': ('Biên bản họp FOMC', 2), 'fed-chair-powell-speech': ('Phát biểu của Chủ tịch Fed', 2), 'fed-chair-powell-testimony': ('Điều trần của Chủ tịch Fed', 1), 'gross-domestic-product-qq': ('GDP (quý)', 2), 'retail-sales-mm': ('Doanh số bán lẻ', 2), 'retail-sales-ex-autos-mm': ('Doanh số bán lẻ lõi', 2), 'producer-price-index-mm': ('Chỉ số giá sản xuất PPI', 2), 'producer-price-index-ex-food-energy-mm': ('PPI lõi', 2), 'ism-manufacturing-pmi': ('PMI sản xuất ISM', 2), 'ism-non-manufacturing-pmi': ('PMI dịch vụ ISM', 2), 'jolts-job-openings': ('Việc làm trống JOLTS', 2), 'adp-nonfarm-employment-change': ('Việc làm ADP', 2), 'initial-jobless-claims': ('Đơn xin trợ cấp thất nghiệp lần đầu', 2), 'consumer-confidence-index': ('Niềm tin tiêu dùng CB', 2), 'michigan-consumer-sentiment': ('Niềm tin tiêu dùng Michigan', 2), } _THU_TU = {k: i for i, k in enumerate(TIN_VANG)} # EA chưa gửi hệ số (multiplier) của MT5: số việc làm ghi theo nghìn / triệu -> thêm đơn vị theo mã _DON_VI_MA = {'nonfarm-payrolls': ' nghìn', 'adp-nonfarm-employment-change': ' nghìn', 'initial-jobless-claims': ' nghìn', 'jolts-job-openings': ' triệu'} SO_LAN_XET = 12 # số lần ra tin gần nhất dùng để tính phản ứng def _vn(t_utc: str) -> datetime: return datetime.strptime(t_utc, '%Y-%m-%d %H:%M:%S') + timedelta(hours=7) def _dv(x: Dict[str, Any]) -> str: return '%' if x.get('don_vi') == 'CALENDAR_UNIT_PERCENT' else _DON_VI_MA.get(x['ma'], '') def _so(x: Optional[float], don_vi: str = '') -> str: if x is None: return '' s = f'{x:,.3f}'.rstrip('0').rstrip('.') return s.replace(',', '§').replace('.', ',').replace('§', '.') + don_vi def thong_ke_phan_ung(goi: Dict[str, Any], eid: int, truoc_utc: str) -> Optional[Dict[str, Any]]: """Phản ứng của vàng ở SO_LAN_XET lần ra tin gần nhất TRƯỚC truoc_utc (không nhìn trước tương lai).""" ds = [x for x in (goi.get('phan_ung_tin') or {}).get(str(eid), []) if x[0] < truoc_utc][-SO_LAN_XET:] if len(ds) < 4: return None bd = [x[3] for x in ds] ra = {'so_lan': len(ds), 'trung_vi_bien_do_2h': round(statistics.median(bd), 2), 'lon_nhat_2h': round(max(bd), 2)} # số thực tế lệch dự báo -> vàng đi hướng nào (cần >= 5 lần có lệch) cao = [x[4] for x in ds if x[1] is not None and x[2] is not None and x[1] > x[2]] thap = [x[4] for x in ds if x[1] is not None and x[2] is not None and x[1] < x[2]] if len(cao) >= 3: ra['khi_cao_hon_du_bao'] = {'so_lan': len(cao), 'vang_giam': sum(1 for h in cao if h < 0)} if len(thap) >= 3: ra['khi_thap_hon_du_bao'] = {'so_lan': len(thap), 'vang_tang': sum(1 for h in thap if h > 0)} return ra def moc_tin(goi: Dict[str, Any], tu: datetime, den: datetime) -> List[Dict[str, Any]]: """Tin vàng trong [tu, den) giờ VN, gộp theo giờ ra tin. Mỗi mốc: giờ, các tin, tin chính, thống kê phản ứng, và nếu đã ra: số thực tế + phản ứng thật của vàng (dùng cho mục tự chấm).""" moc: Dict[str, List[Dict[str, Any]]] = {} for x in goi.get('lich') or []: if x['ma'] not in TIN_VANG: continue t = _vn(x['t_utc']) if tu <= t < den: moc.setdefault(x['t_utc'], []).append(x) ra = [] for t_utc in sorted(moc): ds = sorted(moc[t_utc], key=lambda x: _THU_TU[x['ma']]) chinh = ds[0] t = _vn(t_utc) m = {'ngay': t.strftime('%d/%m'), 'gio': t.strftime('%H:%M'), 'thu': t.weekday(), 'hang': TIN_VANG[chinh['ma']][1], 'tin': [{'ten': TIN_VANG[x['ma']][0], 'du_bao': _so(x['du_bao'], _dv(x)), 'ky_truoc': _so(x['truoc'], _dv(x)), 'thuc_te': _so(x['thuc_te'], _dv(x))} for x in ds], 'phan_ung_cac_lan_truoc': thong_ke_phan_ung(goi, chinh['eid'], t_utc)} that = next((p for p in (goi.get('phan_ung_tin') or {}).get(str(chinh['eid']), []) if p[0] == t_utc), None) if that: m['phan_ung_that'] = {'bien_do_2h': that[3], 'huong_2h': that[4]} ra.append(m) return ra def tin_cho_ban_tin(goi: Dict[str, Any], ngay: date) -> Dict[str, Any]: """Thứ Hai: tin cả tuần; ngày khác: tin từ 7:00 hôm nay tới 7:00 mai (giờ VN).""" bd = datetime.combine(ngay, datetime.min.time()) + timedelta(hours=7) if ngay.weekday() == 0: return {'pham_vi': 'tuan', 'tin': moc_tin(goi, bd, bd + timedelta(days=6))} return {'pham_vi': 'ngay', 'tin': moc_tin(goi, bd, bd + timedelta(days=1))} if __name__ == '__main__': g = {'lich': [{'eid': 1, 'ma': 'nonfarm-payrolls', 't_utc': '2026-10-02 12:30:00', 'thuc_te': 29.0, 'du_bao': 52.0, 'truoc': 22.0, 'muc': 3, 'don_vi': 'CALENDAR_UNIT_JOB'}, {'eid': 2, 'ma': 'unemployment-rate', 't_utc': '2026-10-02 12:30:00', 'thuc_te': 4.2, 'du_bao': 4.1, 'truoc': 4.1, 'muc': 3}, {'eid': 3, 'ma': 'eia-crude-oil-stocks-change', 't_utc': '2026-10-02 14:30:00', 'muc': 3, 'thuc_te': None, 'du_bao': None, 'truoc': None}], 'phan_ung_tin': {'1': [['2026-0%d-05 12:30:00' % i, 50 + i, 40, 20.0 + i, (-1) ** i * 5] for i in range(1, 8)] + [['2026-10-02 12:30:00', 29.0, 52.0, 31.5, 12.0]]}} r = moc_tin(g, datetime(2026, 10, 2, 7), datetime(2026, 10, 3, 7)) assert len(r) == 1 and r[0]['gio'] == '19:30' and [t['ten'] for t in r[0]['tin']][0].startswith('Bảng lương'), r p = r[0]['phan_ung_cac_lan_truoc'] assert p['so_lan'] == 7 and p['trung_vi_bien_do_2h'] == 24.0 and p['khi_cao_hon_du_bao']['so_lan'] == 7, p assert r[0]['phan_ung_that'] == {'bien_do_2h': 31.5, 'huong_2h': 12.0} and r[0]['tin'][0]['thuc_te'] == '29 nghìn' print('TỰ KIỂM ban_tin/lich_tin.py QUA.')