# -*- coding: utf-8 -*- """PHAN_TICH.PY — số liệu bản tin vàng (phần A), tính bằng code từ gói api1 (ban_tin_api1/xuat_so_lieu.py). Gói: vàng 'd1' / 'h4' / 'h1' [[ts, o, h, l, c]…] (ts GIỜ SÀN EET/EEST), 'usdx' {d1, h1}, 'us10y' [[ngày, %]…], 'lich' + 'phan_ung_tin' (lịch MT5), 'regime' (REGIME_DAILY). Nến D1 ngày D đóng lúc 00:00 giờ sàn D+1 = 04:00 (hè) / 05:00 (đông) sáng giờ VN, nên bản tin sáng ngày N nói về nến D1 gần nhất có ngày < N. Bản 2 (02/10/2026, chủ dự án: "lấy cả 11 phần, độ phù hợp ghi theo số, có thêm liên thị trường, gọi tên chung"): 1 tóm tắt (mô hình viết) · 2 đa khung D1/H4/H1 · 3 vùng hỗ trợ/kháng cự (gộp mức trùng thành vùng) · 4 biến động + biên độ dự kiến · 5 động lượng + phân kỳ · 6 liên thị trường (USDX, lợi suất Mỹ 10 năm) · 7 lịch tin + phản ứng lịch sử · 8 ba kịch bản (điều kiện kích hoạt, mục tiêu, mức vô hiệu) · 9 điểm phù hợp Trend / DCA 0–100 · 10 hệ thống hôm nay · 11 tự chấm bản tin hôm qua. LUẬT CỐ ĐỊNH, ngày nào cũng như nhau -> nhất quán và kiểm chứng được. Không dự đoán giá; kịch bản là điều kiện. Không nhìn trước tương lai: chỉ dùng nến đã ĐÓNG trước giờ cắt (diễn tập trên ngày cũ cho kết quả như chạy thật). """ from __future__ import annotations import math import statistics from datetime import date, datetime, timedelta from typing import Any, Dict, List, Optional, Tuple TEN_TRANG_THAI = { 'RALLY': ('Tăng mạnh', 'xu hướng tăng rõ, giá trên đường trung bình và đường trung bình dốc lên'), 'SELLOFF': ('Bán tháo', 'lực bán chiếm ưu thế, giá rơi khỏi vùng cân bằng'), 'CRASH': ('Giảm mạnh', 'xu hướng giảm sâu, giá nằm dưới đường trung bình và đường trung bình dốc xuống'), 'SIDEWAY': ('Đi ngang', 'giá dao động trong biên, chưa có xu hướng rõ'), } # --------------------------------------------------------------------- # GIỜ SÀN <-> GIỜ VN # --------------------------------------------------------------------- def _chu_nhat_cuoi(nam: int, thang: int) -> date: d = date(nam, thang + 1, 1) - timedelta(days=1) if thang < 12 else date(nam, 12, 31) return d - timedelta(days=(d.weekday() + 1) % 7) def lech_san_utc(t_utc: datetime) -> int: """Sàn chạy giờ EET/EEST (theo giờ mùa hè châu Âu): +3 từ CN cuối tháng 3 (01:00 UTC) tới CN cuối tháng 10.""" bd = datetime.combine(_chu_nhat_cuoi(t_utc.year, 3), datetime.min.time()) + timedelta(hours=1) kt = datetime.combine(_chu_nhat_cuoi(t_utc.year, 10), datetime.min.time()) + timedelta(hours=1) return 3 if bd <= t_utc < kt else 2 def san_sang_vn(ts_san: str) -> datetime: t = datetime.strptime(ts_san, '%Y-%m-%d %H:%M') return t - timedelta(hours=lech_san_utc(t - timedelta(hours=3))) + timedelta(hours=7) def ngay_giao_dich_truoc(ngay: date) -> date: d = ngay - timedelta(days=1) while d.weekday() >= 5: d -= timedelta(days=1) return d # --------------------------------------------------------------------- # CHỈ SỐ # --------------------------------------------------------------------- def _tr(nen: List[List[Any]]) -> List[float]: return [max(b[2] - b[3], abs(b[2] - a[4]), abs(b[3] - a[4])) for a, b in zip(nen[:-1], nen[1:])] def _atr(d1: List[List[Any]], n: int = 14) -> Optional[float]: if len(d1) < n + 1: return None return sum(_tr(d1)[-n:]) / n def atr_ds(nen: List[List[Any]], n: int = 14) -> List[float]: """ATR (trung bình TR n phiên) tại từng nến, từ nến thứ n+1.""" tr = _tr(nen) return [sum(tr[i - n:i]) / n for i in range(n, len(tr) + 1)] def ema(gia: List[float], n: int) -> List[Optional[float]]: k, ra, e = 2 / (n + 1), [], None for i, g in enumerate(gia): e = g if e is None else g * k + e * (1 - k) ra.append(e if i >= n - 1 else None) return ra def rsi(gia: List[float], n: int = 14) -> List[Optional[float]]: ra: List[Optional[float]] = [None] * len(gia) if len(gia) <= n: return ra tang = [max(0.0, b - a) for a, b in zip(gia[:-1], gia[1:])] giam = [max(0.0, a - b) for a, b in zip(gia[:-1], gia[1:])] at, ag = sum(tang[:n]) / n, sum(giam[:n]) / n for i in range(n, len(gia)): if i > n: at = (at * (n - 1) + tang[i - 1]) / n ag = (ag * (n - 1) + giam[i - 1]) / n ra[i] = 100.0 if ag == 0 else 100 - 100 / (1 + at / ag) return ra def adx_di(nen: List[List[Any]], n: int = 14) -> List[Tuple[float, float, float]]: """[(ADX, +DI, -DI)] theo Wilder, từ nến thứ 2n.""" if len(nen) < 2 * n + 1: return [] tr = _tr(nen) pdm = [max(b[2] - a[2], 0) if b[2] - a[2] > a[3] - b[3] else 0 for a, b in zip(nen[:-1], nen[1:])] mdm = [max(a[3] - b[3], 0) if a[3] - b[3] > b[2] - a[2] else 0 for a, b in zip(nen[:-1], nen[1:])] st, sp, sm = sum(tr[:n]), sum(pdm[:n]), sum(mdm[:n]) dx, ra, adx = [], [], None for i in range(n, len(tr) + 1): if i > n: st, sp, sm = st - st / n + tr[i - 1], sp - sp / n + pdm[i - 1], sm - sm / n + mdm[i - 1] p, m = (100 * sp / st, 100 * sm / st) if st else (0.0, 0.0) dx.append(100 * abs(p - m) / (p + m) if p + m else 0.0) if len(dx) == n: adx = sum(dx) / n elif len(dx) > n: adx = (adx * (n - 1) + dx[-1]) / n if adx is not None: ra.append((adx, p, m)) return ra def macd(gia: List[float]) -> List[Tuple[float, float]]: """[(MACD, histogram)] 12/26/9.""" e12, e26 = ema(gia, 12), ema(gia, 26) dong = [a - b for a, b in zip(e12, e26) if a is not None and b is not None] tin_hieu = ema(dong, 9) return [(d, d - s) for d, s in zip(dong, tin_hieu) if s is not None] def dinh_day(nen: List[List[Any]], k: int) -> List[Tuple[str, int, float]]: """Đỉnh / đáy sóng (fractal k nến mỗi bên, đã xác nhận): [('H'|'L', chỉ số, giá)].""" ra = [] for i in range(k, len(nen) - k): cao, thap = nen[i][2], nen[i][3] if cao > max(b[2] for b in nen[i - k:i]) and cao >= max(b[2] for b in nen[i + 1:i + k + 1]): ra.append(('H', i, cao)) if thap < min(b[3] for b in nen[i - k:i]) and thap <= min(b[3] for b in nen[i + 1:i + k + 1]): ra.append(('L', i, thap)) return ra def moc_tron(gia: float, buoc: float = 50) -> Dict[str, float]: return {'duoi': math.floor(gia / buoc) * buoc, 'tren': math.ceil(gia / buoc) * buoc if gia % buoc else gia + buoc} def _tuong_quan(a: List[float], b: List[float]) -> Optional[float]: if len(a) < 10 or len(a) != len(b): return None try: return statistics.correlation(a, b) except statistics.StatisticsError: return None def _r(x: Optional[float], le: int = 2) -> Optional[float]: return None if x is None else round(x, le) # --------------------------------------------------------------------- # 2. ĐA KHUNG # --------------------------------------------------------------------- def khung(nen: List[List[Any]], ten: str) -> Dict[str, Any]: dong = [b[4] for b in nen] e50, e200 = ema(dong, 50), ema(dong, 200) gia, diem = dong[-1], 0 if e50[-1] is not None: diem += 1 if gia > e50[-1] else -1 if e50[-1] is not None and e200[-1] is not None: diem += 1 if e50[-1] > e200[-1] else -1 dd = dinh_day(nen[-80:], 2 if ten == 'D1' else 3) h = [x[2] for x in dd if x[0] == 'H'][-2:] l_ = [x[2] for x in dd if x[0] == 'L'][-2:] cau_truc = 'đan xen' if len(h) == 2 and len(l_) == 2: if h[1] > h[0] and l_[1] > l_[0]: cau_truc, diem = 'đỉnh và đáy cao dần', diem + 1 elif h[1] < h[0] and l_[1] < l_[0]: cau_truc, diem = 'đỉnh và đáy thấp dần', diem - 1 nhan = 'Tăng' if diem >= 2 else ('Giảm' if diem <= -2 else 'Giằng co') return {'khung': ten, 'nhan': nhan, 'diem': diem, 'gia': _r(gia), 'ema50': _r(e50[-1]), 'ema200': _r(e200[-1]), 'tren_ema50': None if e50[-1] is None else gia > e50[-1], 'tren_ema200': None if e200[-1] is None else gia > e200[-1], 'cau_truc': cau_truc} # --------------------------------------------------------------------- # 3. VÙNG GIÁ # --------------------------------------------------------------------- def cac_muc(d1: List[List[Any]], h4: List[List[Any]], h1_hom_nay: List[List[Any]], ngay: date, atr: float) -> List[Dict[str, Any]]: """Mọi mức giá kỹ thuật + trọng số (mức càng nhiều phương pháp cùng chỉ ra càng mạnh).""" hq = d1[-1] cao, thap, dong = hq[2], hq[3], hq[4] p = (cao + thap + dong) / 3 muc = [('Pivot ngày', p, 1), ('R1 pivot', 2 * p - thap, 1), ('S1 pivot', 2 * p - cao, 1), ('R2 pivot', p + (cao - thap), .75), ('S2 pivot', p - (cao - thap), .75), ('Đỉnh hôm qua', cao, 1.5), ('Đáy hôm qua', thap, 1.5)] dau_tuan = ngay - timedelta(days=ngay.weekday()) tuan_truoc = [b for b in d1 if dau_tuan - timedelta(days=7) <= date.fromisoformat(b[0][:10]) < dau_tuan] if tuan_truoc: muc += [('Đỉnh tuần trước', max(b[2] for b in tuan_truoc), 2), ('Đáy tuần trước', min(b[3] for b in tuan_truoc), 2)] tuan_nay = [b for b in d1 if date.fromisoformat(b[0][:10]) >= dau_tuan] mo_tuan = tuan_nay[0][1] if tuan_nay else (h1_hom_nay[0][1] if h1_hom_nay else None) if mo_tuan: muc.append(('Giá mở cửa tuần', mo_tuan, 1)) thang = [b for b in d1 if b[0][:7] == ngay.isoformat()[:7]] mo_thang = thang[0][1] if thang else (h1_hom_nay[0][1] if h1_hom_nay else None) if mo_thang: muc.append(('Giá mở cửa tháng', mo_thang, 1)) for x in dinh_day(d1[-60:], 2): muc.append(('Đỉnh sóng D1' if x[0] == 'H' else 'Đáy sóng D1', x[2], 2)) for x in dinh_day(h4[-90:], 3): muc.append(('Đỉnh sóng H4' if x[0] == 'H' else 'Đáy sóng H4', x[2], 1.5)) # Fibonacci của sóng D1 gần nhất (đủ lớn: >= 1,5 ATR) dd = dinh_day(d1[-40:], 2) hs, ls = [x for x in dd if x[0] == 'H'], [x for x in dd if x[0] == 'L'] if hs and ls: h, l_ = hs[-1], ls[-1] if h[2] - l_[2] >= 1.5 * atr: for r in (.382, .5, .618): g = h[2] - r * (h[2] - l_[2]) if h[1] > l_[1] else l_[2] + r * (h[2] - l_[2]) muc.append((f'Fibonacci {r * 100:.1f}%'.replace('.', ','), g, 1)) gia = h1_hom_nay[-1][4] if h1_hom_nay else dong for buoc in (50, 100): mt = moc_tron(gia, buoc) for g in (mt['duoi'], mt['tren'], mt['duoi'] - buoc, mt['tren'] + buoc): muc.append((f'Mốc tròn {int(g):,}'.replace(',', '.'), float(g), 1.25 if g % 100 == 0 else .75)) return [{'ten': t, 'gia': g, 'trong_so': w} for t, g, w in muc] def gom_vung(muc: List[Dict[str, Any]], atr: float, gia: float, xa: float = 3.0) -> Dict[str, List[Dict[str, Any]]]: """Gộp các mức cách nhau <= 0,25 ATR thành vùng; trả 3 vùng hỗ trợ (dưới giá) và 3 kháng cự (trên giá) gần nhất.""" ds = sorted((m for m in muc if abs(m['gia'] - gia) <= xa * atr), key=lambda m: m['gia']) nhom: List[List[Dict[str, Any]]] = [] for m in ds: if nhom and m['gia'] - nhom[-1][0]['gia'] <= .25 * atr: nhom[-1].append(m) else: nhom.append([m]) vung = [] for n in nhom: ten = [] for m in n: if m['ten'] not in ten: ten.append(m['ten']) diem = sum(m['trong_so'] for m in n) vung.append({'duoi': round(min(m['gia'] for m in n), 2), 'tren': round(max(m['gia'] for m in n), 2), 'giua': round(sum(m['gia'] for m in n) / len(n), 2), 'diem': round(diem, 2), 'do_manh': 'mạnh' if diem >= 4 else ('vừa' if diem >= 2.5 else 'yếu'), 'gom': ten}) # vùng yếu một mức (vd chỉ một mốc tròn 50) bỏ nếu đã đủ vùng ho_tro = sorted([v for v in vung if v['giua'] < gia], key=lambda v: -v['giua']) khang_cu = sorted([v for v in vung if v['giua'] >= gia], key=lambda v: v['giua']) def _chon(ds_v): manh = [v for v in ds_v if v['diem'] >= 2.5] return (manh if len(manh) >= 2 else ds_v)[:3] return {'ho_tro': _chon(ho_tro), 'khang_cu': _chon(khang_cu)} # --------------------------------------------------------------------- # BỐI CẢNH LỚN: THIÊN HƯỚNG THÁNG / TUẦN (phân tích từ trên xuống) # --------------------------------------------------------------------- # 03/10/2026 chủ dự án: "về chuyên nghiệp phải cần có 1 phân tích về tầm nhìn tuần và tháng, sau đó phân tích theo # ngày". Thiên hướng chỉ dùng tháng / tuần ĐÃ ĐÓNG (không nhìn trước), luật cố định: # tháng = điểm xu hướng khung tuần (giá vs EMA50/200 tuần, cấu trúc đỉnh đáy, -3..+3) + nến tháng vừa đóng (+1 / -1) # tuần = điểm xu hướng khung ngày (-3..+3) + nến tuần vừa đóng (+1 / -1) # >= +2 Tăng, <= -2 Giảm, còn lại Trung tính. Vô hiệu: đóng tuần (tháng) / đóng ngày (tuần) vượt đỉnh-đáy kỳ trước. def nen_thang(d1: List[List[Any]]) -> List[List[Any]]: ra: List[List[Any]] = [] for b in d1: dau = b[0][:7] + '-01' if ra and ra[-1][0][:10] == dau: ra[-1][2], ra[-1][3], ra[-1][4] = max(ra[-1][2], b[2]), min(ra[-1][3], b[3]), b[4] else: ra.append([dau + ' 00:00', b[1], b[2], b[3], b[4]]) return ra def _nhan_huong(diem: int) -> str: return 'Tăng' if diem >= 2 else ('Giảm' if diem <= -2 else 'Trung tính') def thien_huong(nen_lon: List[List[Any]], nen_nho: List[List[Any]], ten_nho: str, ky: str) -> Optional[Dict[str, Any]]: """nen_lon: nến kỳ (tháng / tuần) ĐÃ ĐÓNG; nen_nho: nến khung dưới để chấm xu hướng (tuần / ngày).""" if len(nen_lon) < 2 or len(nen_nho) < 60: return None k = khung(nen_nho, ten_nho) cuoi, truoc = nen_lon[-1], nen_lon[-2] nen = 1 if cuoi[4] > cuoi[1] else -1 diem = k['diem'] + nen huong = _nhan_huong(diem) kq = {'huong': huong, 'diem': diem, 'xu_huong_khung_duoi': k['nhan'], 'cau_truc': k['cau_truc'], 'nen_ky_truoc': 'tăng' if nen > 0 else 'giảm', 'ky_truoc': {'mo': cuoi[1], 'cao': cuoi[2], 'thap': cuoi[3], 'dong': cuoi[4]}} don_vi = 'tuần' if ky == 'tháng' else 'ngày' if huong == 'Tăng': kq['vo_hieu'] = f'đóng {don_vi} dưới đáy {ky} trước {_gvn(cuoi[3])}' elif huong == 'Giảm': kq['vo_hieu'] = f'đóng {don_vi} trên đỉnh {ky} trước {_gvn(cuoi[2])}' else: kq['vo_hieu'] = f'đóng {don_vi} ra ngoài biên {ky} trước {_gvn(cuoi[3])}–{_gvn(cuoi[2])} thì mới có hướng' return kq def boi_canh(goi: Dict[str, Any], ngay: date, gia: float) -> Dict[str, Any]: d1 = [b for b in goi['d1'] if b[0][:10] < ngay.isoformat()] dau_thang = ngay.replace(day=1).isoformat() dau_tuan = (ngay - timedelta(days=ngay.weekday())).isoformat() d1_thang = [b for b in d1 if b[0][:10] < dau_thang] d1_tuan = [b for b in d1 if b[0][:10] < dau_tuan] th = thien_huong(nen_thang(d1_thang), nen_tuan(d1_thang), 'W1', 'tháng') tu = thien_huong(nen_tuan(d1_tuan), d1_tuan, 'D1', 'tuần') ra: Dict[str, Any] = {'thang': th, 'tuan': tu} for k, x in (('thang', th), ('tuan', tu)): if x: c, t = x['ky_truoc']['cao'], x['ky_truoc']['thap'] x['vi_tri_gia'] = ('trên đỉnh' if gia > c else 'dưới đáy' if gia < t else f'trong biên, ở {round((gia - t) / (c - t) * 100)}% từ đáy lên') + (' kỳ trước') nam = [b for b in d1 if b[0][:10] >= (ngay - timedelta(days=365)).isoformat()] if nam: c, t = max(b[2] for b in nam), min(b[3] for b in nam) ra['nam'] = {'dinh_52_tuan': c, 'day_52_tuan': t, 'vi_tri_pct': round((gia - t) / (c - t) * 100) if c > t else None} return ra # --------------------------------------------------------------------- # 9. ĐIỂM PHÙ HỢP TREND / DCA (0–100) # --------------------------------------------------------------------- def _kep(x: float, a: float = 0.0, b: float = 1.0) -> float: return max(a, min(b, x)) def _so(x: float, le: int = 1) -> str: return f'{x:,.{le}f}'.replace(',', '§').replace('.', ',').replace('§', '.') def diem_trend(dl: Dict[str, Any]) -> Dict[str, Any]: """thanh_phan: (lời thường, kèm chỉ số, điểm) — bản lời thường của bản tin không nêu tên chỉ báo.""" adx, pdi, mdi = dl['adx'], dl['pdi'], dl['mdi'] huong_d1 = dl['khung'][0]['nhan'] huong = huong_d1 if huong_d1 != 'Giằng co' else ('Tăng' if pdi > mdi else 'Giảm') cung = sum(1 for k in dl['khung'] if k['nhan'] == huong) xa = dl['xa_ema50_atr'] ten_tt = TEN_TRANG_THAI.get(dl['regime'] or '', (dl['regime'] or 'chưa có',))[0] phan = [ ('độ mạnh xu hướng khung ngày', f'ADX D1 {_so(adx)}', 30 * _kep((adx - 15) / 20)), (f"lực {'bán' if mdi > pdi else 'mua'} lấn át", f'chênh lệch +DI/−DI {_so(abs(pdi - mdi))}', 15 * _kep(abs(pdi - mdi) / 20)), (f'{cung}/{len(dl["khung"])} khung thời gian cùng hướng {huong.lower()}', f'{cung} khung cùng hướng', 10 * cung), (f"giá cách đường trung bình 50 ngày {_so(abs(xa), 2)} lần biên độ ngày", f'giá cách EMA50 D1 {_so(xa, 2)} ATR', 10 * _kep(abs(xa) / 1.5)), (f'trạng thái hệ thống: {ten_tt}', f'trạng thái hệ thống: {ten_tt}', 15 if dl['regime'] in ('RALLY', 'CRASH', 'SELLOFF') else 0), ] diem = round(min(100, sum(p[2] for p in phan))) return {'diem': diem, 'huong': huong, 'muc': _muc(diem), 'thanh_phan': [{'yeu_to': t, 'chi_so': c, 'diem': round(d, 1)} for t, c, d in phan]} def diem_dca(dl: Dict[str, Any]) -> Dict[str, Any]: xa = dl['xa_ema50_atr'] ten_tt = TEN_TRANG_THAI.get(dl['regime'] or '', (dl['regime'] or 'chưa có',))[0] tru = [ ('xu hướng một chiều đang mạnh lên', f"ADX D1 {_so(dl['adx'])}", 30 * _kep((dl['adx'] - 18) / 17)), (f"biến động cao so với 3 tháng (cao hơn {dl['atr_bach_phan_vi']:.0f}% số ngày)", f"ATR bách phân vị {dl['atr_bach_phan_vi']:.0f}", 20 * _kep((dl['atr_bach_phan_vi'] - 60) / 40)), ('giá chạy xa đường trung bình 50 ngày', f'giá cách EMA50 D1 {_so(xa, 2)} ATR', 15 * _kep((abs(xa) - 1) / 2)), ('cả 3 khung thời gian cùng một hướng', 'cả 3 khung cùng hướng', 10 if len({k['nhan'] for k in dl['khung']}) == 1 and dl['khung'][0]['nhan'] != 'Giằng co' else 0), (f'trạng thái hệ thống: {ten_tt}', f'trạng thái hệ thống: {ten_tt}', {'CRASH': 15, 'SELLOFF': 10, 'RALLY': 10}.get(dl['regime'], 0)), ('có tin mạnh trong ngày', 'có tin mạnh trong ngày', {1: 20, 2: 10}.get(dl['tin_manh_nhat'], 0)), ] diem = round(max(0, 100 - sum(p[2] for p in tru))) return {'diem': diem, 'muc': _muc(diem), 'tru_diem': [{'yeu_to': t, 'chi_so': c, 'tru': round(d, 1)} for t, c, d in tru if d > 0]} def _muc(diem: int) -> str: return 'Thuận lợi' if diem >= 70 else ('Trung bình' if diem >= 45 else 'Bất lợi') def loi_khuyen(trend: Dict[str, Any], dca: Dict[str, Any], kb: Dict[str, Any], tin: List[Dict[str, Any]]) -> Dict[str, str]: """Câu khuyên QUẢN TRỊ (không phải lệnh mua/bán) theo dải điểm — câu cố định trong code.""" h = trend['huong'].lower() vo_hieu = kb['tang']['vo_hieu'] if trend['huong'] == 'Tăng' else kb['giam']['vo_hieu'] t = {'Thuận lợi': f'Xu hướng {h} đủ mạnh: ưu tiên đi theo hướng {h}, không bắt đỉnh/đáy ngược xu hướng; xu hướng mất ' f'hiệu lực nếu giá {vo_hieu}.', 'Trung bình': f'Xu hướng {h} chưa đủ mạnh: giảm khối lượng, chỉ theo khi giá phá vùng có xác nhận (nến H1 đóng ' f'ngoài vùng).', 'Bất lợi': 'Không có xu hướng rõ: chiến lược theo xu hướng dễ bị quét hai đầu, nên đứng ngoài hoặc chờ phá vỡ.' }[trend['muc']] d = {'Thuận lợi': 'Điều kiện phù hợp với DCA: chạy theo cấu hình đã kiểm thử, vẫn giữ giới hạn lỗ tối đa (MaxLoss).', 'Trung bình': 'Nên giãn bước giá giữa các tầng, giảm lot khởi đầu, giới hạn số tầng.', 'Bất lợi': 'Rủi ro cao cho DCA: giãn bước giá mạnh, giảm lot, giới hạn số tầng, tôn trọng MaxLoss; cân nhắc ' 'tạm dừng mở chuỗi mới.'}[dca['muc']] manh = [m for m in tin if m['hang'] == 1] if manh: m = manh[0] g = datetime.strptime(m['gio'], '%H:%M') d += (f" Quanh tin {m['tin'][0]['ten']} lúc {m['gio']}: không mở chuỗi mới từ " f"{(g - timedelta(minutes=30)).strftime('%H:%M')} đến {(g + timedelta(minutes=90)).strftime('%H:%M')}.") return {'trend': t, 'dca': d} # --------------------------------------------------------------------- # 8. KỊCH BẢN # --------------------------------------------------------------------- def _gvn(x: float) -> str: return f'{x:,.2f}'.replace(',', '§').replace('.', ',').replace('§', '.') def kich_ban(vung: Dict[str, List[Dict[str, Any]]], gia: float, atr: float, bc: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """Ba kịch bản dạng điều kiện. 03/10/2026 chủ dự án chọn BỎ "kịch bản ưu tiên": kiểm trên 54 phiên, mọi luật chọn hướng cho một ngày đều thua việc không chọn (đúng 7/34 và 5/34 ngày có chọn, trong khi 56% số ngày giá không chạm mục tiêu nào). Thay bằng tần suất thật (tan_suat_kich_ban) + đánh dấu kịch bản thuận bối cảnh tuần / tháng. Mức kích hoạt = mép vùng gần nhất cách giá 0,2–1,2 ATR (ưu tiên vùng không yếu); không có thì giá ± 0,5 ATR. Mục tiêu = vùng kế tiếp; vô hiệu = mức kích hoạt phía bên kia.""" kc = [v for v in vung['khang_cu'] if .2 * atr <= v['tren'] - gia <= 1.2 * atr] ht = [v for v in vung['ho_tro'] if .2 * atr <= gia - v['duoi'] <= 1.2 * atr] kc_c = next((v for v in kc if v['do_manh'] != 'yếu'), kc[0] if kc else None) ht_c = next((v for v in ht if v['do_manh'] != 'yếu'), ht[0] if ht else None) r1 = kc_c['tren'] if kc_c else round(gia + .5 * atr, 2) s1 = ht_c['duoi'] if ht_c else round(gia - .5 * atr, 2) sau_r = [v for v in vung['khang_cu'] if v['giua'] > r1 + .2 * atr] sau_s = [v for v in vung['ho_tro'] if v['giua'] < s1 - .2 * atr] r2 = sau_r[0]['giua'] if sau_r else round(r1 + .5 * atr, 2) s2 = sau_s[0]['giua'] if sau_s else round(s1 - .5 * atr, 2) tu = ((bc or {}).get('tuan') or {}).get('huong') th = ((bc or {}).get('thang') or {}).get('huong') thuan = ('tang' if tu == 'Tăng' and th != 'Giảm' else 'giam' if tu == 'Giảm' and th != 'Tăng' else 'di_ngang' if tu == 'Trung tính' else None) return { 'tang': {'kich_hoat': r1, 'muc_tieu': r2, 'vo_hieu_gia': s1, 'vo_hieu': f'đóng nến H1 dưới {_gvn(s1)}'}, 'giam': {'kich_hoat': s1, 'muc_tieu': s2, 'vo_hieu_gia': r1, 'vo_hieu': f'đóng nến H1 trên {_gvn(r1)}'}, 'di_ngang': {'tu': s1, 'den': r1}, 'thuan_boi_canh': thuan, } # --------------------------------------------------------------------- # GÓI SỐ LIỆU CHO MỘT BUỔI SÁNG # --------------------------------------------------------------------- def _cat_nen(nen: List[List[Any]], cat: datetime, gio: int) -> List[List[Any]]: """Nến (giờ sàn) đã ĐÓNG trước giờ cắt (giờ VN).""" return [b for b in nen if san_sang_vn(b[0]) + timedelta(hours=gio) <= cat] def so_lieu_sang(goi: Dict[str, Any], ngay: date, gio_cat: int = 7, tu_cham: bool = True) -> Dict[str, Any]: """Số liệu bản tin sáng ngày `ngay` (giờ VN), chỉ dùng dữ liệu có trước gio_cat giờ sáng hôm đó.""" from ban_tin.lich_tin import tin_cho_ban_tin cat = datetime.combine(ngay, datetime.min.time()) + timedelta(hours=gio_cat) d1 = [b for b in goi['d1'] if b[0][:10] < ngay.isoformat()] if len(d1) < 60: raise ValueError('Thiếu nến D1.') h4 = _cat_nen(goi.get('h4') or [], cat, 4) h1_san = _cat_nen(goi['h1'], cat, 1) hom_qua, truoc = d1[-1], d1[-2] ngay_nen = hom_qua[0][:10] atr_list = atr_ds(d1) atr14 = atr_list[-1] bien_do = hom_qua[2] - hom_qua[3] # 'hôm nay' theo đồng hồ sàn (nến D1 hôm nay đã mở lúc 04:00/05:00 VN) h1_hom_nay = [b for b in h1_san if b[0][:10] == ngay.isoformat()] h1 = [] for b in h1_san: vn = san_sang_vn(b[0]) if vn >= cat - timedelta(days=6): h1.append([vn.strftime('%Y-%m-%d %H:%M')] + b[1:]) gia_moi = h1[-1][4] if h1 else hom_qua[4] # --- 2 đa khung cac_khung = [khung(d1, 'D1')] if len(h4) >= 60: cac_khung.append(khung(h4, 'H4')) if len(h1_san) >= 60: cac_khung.append(khung(h1_san, 'H1')) # --- 3 vùng vung = gom_vung(cac_muc(d1, h4, h1_hom_nay, ngay, atr14), atr14, gia_moi) # --- 4 biến động gan = atr_list[-63:] bach_phan_vi = 100 * sum(1 for a in gan if a <= atr14) / len(gan) bien_dong = {'atr14': round(atr14, 2), 'atr_bach_phan_vi_3_thang': round(bach_phan_vi), 'atr_trung_vi_3_thang': round(statistics.median(gan), 2)} if h1_hom_nay: c, t = max(b[2] for b in h1_hom_nay), min(b[3] for b in h1_hom_nay) bien_dong.update({'mo_cua_hom_nay': h1_hom_nay[0][1], 'cao_tu_dau_ngay': c, 'thap_tu_dau_ngay': t, 'da_chay': round(c - t, 2), 'da_chay_pct_atr': round((c - t) / atr14 * 100), 'tran_du_kien': round(t + atr14, 2), 'san_du_kien': round(c - atr14, 2)}) # --- 5 động lượng dong_d1 = [b[4] for b in d1] r_d1 = rsi(dong_d1)[-1] r_h4 = rsi([b[4] for b in h4])[-1] if len(h4) > 20 else None md = macd(dong_d1) ad = adx_di(d1) adx, pdi, mdi = ad[-1] dong_luong = {'rsi_d1': _r(r_d1, 1), 'rsi_h4': _r(r_h4, 1), 'adx_d1': round(adx, 1), 'adx_3_phien_truoc': round(ad[-4][0], 1), 'pdi': round(pdi, 1), 'mdi': round(mdi, 1), 'macd_d1': _r(md[-1][0]), 'macd_hist_d1': _r(md[-1][1]), 'macd_hist_truoc': _r(md[-2][1]), 'phan_ky_h4': phan_ky(h4)} # --- 6 liên thị trường lien = lien_thi_truong(goi, d1, ngay) # --- 7 lịch tin lich = tin_cho_ban_tin(goi, ngay) tin_hom_nay = lich['tin'] if lich['pham_vi'] == 'ngay' else [m for m in lich['tin'] if m['ngay'] == ngay.strftime('%d/%m')] # --- 9 điểm + 8 kịch bản reg = {r['date']: r for r in goi.get('regime') or []} tt = reg.get(ngay_nen) truoc_tt = None for r in sorted(reg): if r < ngay_nen: truoc_tt = reg[r] e50 = ema(dong_d1, 50)[-1] dl = {'adx': adx, 'pdi': pdi, 'mdi': mdi, 'khung': cac_khung, 'xa_ema50_atr': (gia_moi - e50) / atr14, 'regime': tt['regime'] if tt else None, 'atr_bach_phan_vi': bach_phan_vi, 'tin_manh_nhat': min((m['hang'] for m in tin_hom_nay), default=None)} trend, dca = diem_trend(dl), diem_dca(dl) bc = boi_canh(goi, ngay, gia_moi) kb = kich_ban(vung, gia_moi, atr14, bc) trang_thai = None if tt: ten, mo_ta = TEN_TRANG_THAI.get(tt['regime'], (tt['regime'], '')) trang_thai = {'ma': tt['regime'], 'ten': ten, 'mo_ta': mo_ta, 'vua_chuyen': bool(truoc_tt and truoc_tt['regime'] != tt['regime']), 'tu': (TEN_TRANG_THAI.get(truoc_tt['regime'], (truoc_tt['regime'],))[0] if truoc_tt else None), 'adx': round(float(tt['adx']), 1), 'atr_pct': round(float(tt['atr_pct']), 2), 'rsi': round(float(tt['rsi']), 1), 'bb_width': round(float(tt['bb_width']), 2), 'gia_so_tb_pct': round(float(tt['price_vs_sma']), 2), 'doc_tb': round(float(tt['sma_slope']), 3)} dau_tuan = ngay - timedelta(days=ngay.weekday()) tuan_truoc = [b for b in d1 if dau_tuan - timedelta(days=7) <= date.fromisoformat(b[0][:10]) < dau_tuan] tuan_nay = [b for b in d1 if date.fromisoformat(b[0][:10]) >= dau_tuan] thay_doi = hom_qua[4] - truoc[4] atr_hq = atr_list[-2] if len(atr_list) > 1 else atr14 # vùng cho biểu đồ: 2 hỗ trợ + 2 kháng cự gần nhất ve = ([{'ten': f'Kháng cự {i + 1}', 'gia': v['giua'], 'loai': 'kc'} for i, v in enumerate(vung['khang_cu'][:2])] + [{'ten': f'Hỗ trợ {i + 1}', 'gia': v['giua'], 'loai': 'ht'} for i, v in enumerate(vung['ho_tro'][:2])]) ra = { 'ngay_ban_tin': ngay.isoformat(), 'thu': ngay.weekday(), 'ngay_nen': ngay_nen, 'mo_cua_tuan': ngay.weekday() == 0, 'hom_qua': {'mo': hom_qua[1], 'cao': hom_qua[2], 'thap': hom_qua[3], 'dong': hom_qua[4], 'thay_doi': round(thay_doi, 2), 'thay_doi_pct': round(thay_doi / truoc[4] * 100, 2), 'bien_do': round(bien_do, 2), 'bien_do_so_tb': round(bien_do / atr_hq, 2)}, 'atr14': round(atr14, 2), 'tuan_nay': ({'mo': tuan_nay[0][1], 'cao': max(b[2] for b in tuan_nay), 'thap': min(b[3] for b in tuan_nay), 'dong': tuan_nay[-1][4]} if tuan_nay else None), 'tuan_truoc': ({'mo': tuan_truoc[0][1], 'cao': max(b[2] for b in tuan_truoc), 'thap': min(b[3] for b in tuan_truoc), 'dong': tuan_truoc[-1][4]} if tuan_truoc else None), 'gia_sang_nay': round(gia_moi, 2), 'gio_gia': h1[-1][0] if h1 else None, 'trang_thai': trang_thai, 'da_khung': cac_khung, 'vung_gia': vung, 'bien_dong': bien_dong, 'dong_luong': dong_luong, 'lien_thi_truong': lien, 'lich': lich, 'kich_ban': kb, 'boi_canh': bc, 'su_kien_fed_24h': _su_kien(cat - timedelta(days=3 if ngay.weekday() == 0 else 1), cat), 'diem_trend': trend, 'diem_dca': dca, 'loi_khuyen': loi_khuyen(trend, dca, kb, tin_hom_nay), 'vung': ve, 'h1': h1, } if ra['mo_cua_tuan'] and h1_hom_nay: ra['gap'] = round(h1_hom_nay[0][1] - hom_qua[4], 2) # mở cửa tuần so với đóng thứ Sáu if tu_cham: # bài đăng thật: thêm tự chấm + tần suất (bản dựng lại để chấm thì không cần) try: ra['tu_cham'] = tu_cham_hom_qua(goi, ngay, gio_cat) except ValueError: ra['tu_cham'] = None ra['tan_suat'] = tan_suat_kich_ban(goi, ngay, gio_cat=gio_cat) return ra _DEM_TAN_SUAT: Dict[Tuple[int, str], Dict[str, Any]] = {} def tan_suat_kich_ban(goi: Dict[str, Any], ngay: date, so_phien: int = 50, gio_cat: int = 7) -> Dict[str, Any]: """Trong <= so_phien phiên gần nhất trước `ngay`, dựng lại từng bản tin và chấm: bao nhiêu % ngày chạm mục tiêu tăng trước / giảm trước / không chạm mục tiêu nào. Cùng luật kịch bản với bài hôm nay -> số tham khảo trung thực.""" khoa = (id(goi), ngay.isoformat()) if khoa in _DEM_TAN_SUAT: return _DEM_TAN_SUAT[khoa] dem = {'tang': 0, 'giam': 0, 'di_ngang': 0} x, n = ngay, 0 while n < so_phien and x > ngay - timedelta(days=so_phien * 2): if x.weekday() < 5: try: dem[tu_cham_hom_qua(goi, x, gio_cat)['kich_ban_xay_ra']] += 1 n += 1 except (ValueError, IndexError, KeyError): pass x -= timedelta(days=1) ra = {'so_phien': n, **{k: (round(v * 100 / n) if n else None) for k, v in dem.items()}} _DEM_TAN_SUAT[khoa] = ra return ra def _su_kien(tu: datetime, den: datetime) -> List[Dict[str, Any]]: from ban_tin.su_kien import trong return trong(tu, den) def phan_ky(h4: List[List[Any]]) -> Optional[str]: """Phân kỳ RSI trên H4 ở 2 đỉnh / 2 đáy sóng gần nhất (đỉnh sau phải mới trong 12 nến).""" if len(h4) < 60: return None nen = h4[-80:] r = rsi([b[4] for b in h4])[-80:] dd = dinh_day(nen, 3) hs, ls = [x for x in dd if x[0] == 'H'][-2:], [x for x in dd if x[0] == 'L'][-2:] if len(hs) == 2 and hs[1][1] >= len(nen) - 15 and r[hs[0][1]] and r[hs[1][1]]: if hs[1][2] > hs[0][2] and r[hs[1][1]] < r[hs[0][1]] - 2: return 'phân kỳ giảm (giá lập đỉnh cao hơn nhưng RSI thấp hơn)' if len(ls) == 2 and ls[1][1] >= len(nen) - 15 and r[ls[0][1]] and r[ls[1][1]]: if ls[1][2] < ls[0][2] and r[ls[1][1]] > r[ls[0][1]] + 2: return 'phân kỳ tăng (giá lập đáy thấp hơn nhưng RSI cao hơn)' return None def lien_thi_truong(goi: Dict[str, Any], d1: List[List[Any]], ngay: date) -> Dict[str, Any]: ra: Dict[str, Any] = {} ux = [b for b in (goi.get('usdx') or {}).get('d1') or [] if b[0][:10] < ngay.isoformat()] if len(ux) >= 60: dong = [b[4] for b in ux] k = khung(ux, 'D1') ra['usdx'] = {'ngay': ux[-1][0][:10], 'dong': round(dong[-1], 3), 'doi_1_phien_pct': round((dong[-1] / dong[-2] - 1) * 100, 2), 'doi_5_phien_pct': round((dong[-1] / dong[-6] - 1) * 100, 2), 'xu_huong_d1': k['nhan'], 'tren_ema50': k['tren_ema50']} vang = {b[0][:10]: b[4] for b in d1} usd = {b[0][:10]: b[4] for b in ux} chung = sorted(set(vang) & set(usd))[-21:] if len(chung) >= 11: rv = [vang[b] / vang[a] - 1 for a, b in zip(chung[:-1], chung[1:])] ru = [usd[b] / usd[a] - 1 for a, b in zip(chung[:-1], chung[1:])] tq = _tuong_quan(rv, ru) ra['tuong_quan_vang_usdx_20_phien'] = _r(tq) ls = [x for x in goi.get('us10y') or [] if x[0] < ngay.isoformat()] if len(ls) >= 6: ra['loi_suat_my_10_nam'] = {'ngay': ls[-1][0], 'muc_pct': ls[-1][1], 'doi_1_phien_diem_co_ban': round((ls[-1][1] - ls[-2][1]) * 100), 'doi_5_phien_diem_co_ban': round((ls[-1][1] - ls[-6][1]) * 100)} return ra # --------------------------------------------------------------------- # 11. TỰ CHẤM BẢN TIN HÔM QUA # --------------------------------------------------------------------- def tu_cham_hom_qua(goi: Dict[str, Any], ngay: date, gio_cat: int = 7) -> Dict[str, Any]: """Đối chiếu bản tin phiên giao dịch trước (dựng lại đúng như sáng hôm đó) với diễn biến thật tới giờ cắt hôm nay.""" truoc = ngay_giao_dich_truoc(ngay) sl = so_lieu_sang(goi, truoc, gio_cat, tu_cham=False) bd = datetime.combine(truoc, datetime.min.time()) + timedelta(hours=gio_cat) kt = datetime.combine(ngay, datetime.min.time()) + timedelta(hours=gio_cat) duong = [] for b in goi['h1']: vn = san_sang_vn(b[0]) if bd <= vn and vn + timedelta(hours=1) <= kt: duong.append([vn.strftime('%d/%m %H:%M')] + b[1:]) if not duong: raise ValueError('Chưa có nến sau bản tin trước.') kb = sl['kich_ban'] # 03/10/2026 chủ dự án đồng ý: kịch bản xảy ra = kịch bản CHẠM MỤC TIÊU TRƯỚC; không mục tiêu nào chạm = đi ngang. # (Bản đầu: nến H1 đầu tiên đóng qua mức kích hoạt -> giá chỉ chạm rồi quay đầu cũng tính, đi ngang gần như luôn sai.) xay_ra, luc = 'di_ngang', None for b in duong: len_ = b[2] >= kb['tang']['muc_tieu'] xuong = b[3] <= kb['giam']['muc_tieu'] if len_ and xuong: # một nến chạm cả hai: lấy mục tiêu gần giá mở nến hơn (thường chạm trước) len_ = kb['tang']['muc_tieu'] - b[1] <= b[1] - kb['giam']['muc_tieu'] xuong = not len_ if len_ or xuong: xay_ra, luc = ('tang' if len_ else 'giam'), b[0] break kich_hoat, luc_kh = None, None # thông tin thêm: mức kích hoạt nào bị vượt trước (nến H1 đóng qua) for b in duong: if b[4] > kb['tang']['kich_hoat'] or b[4] < kb['giam']['kich_hoat']: kich_hoat, luc_kh = ('tang' if b[4] > kb['tang']['kich_hoat'] else 'giam'), b[0] break cao, thap = max(b[2] for b in duong), min(b[3] for b in duong) vung_cham = [] for loai, ds in (('kháng cự', sl['vung_gia']['khang_cu']), ('hỗ trợ', sl['vung_gia']['ho_tro'])): for i, v in enumerate(ds): if thap <= v['tren'] and cao >= v['duoi']: vung_cham.append(f'{loai} {i + 1} ({_gvn(v["duoi"])}–{_gvn(v["tren"])})') bdk = sl['bien_dong'] tin_that = [m for m in sl['lich']['tin'] if 'phan_ung_that' in m and (sl['lich']['pham_vi'] == 'ngay' or m['ngay'] == truoc.strftime('%d/%m'))] return { 'ban_tin_ngay': truoc.isoformat(), 'kich_ban_xay_ra': xay_ra, 'luc_cham_muc_tieu': luc, 'thuan_boi_canh': kb.get('thuan_boi_canh'), 'muc_tieu_hom_qua': {'tang': kb['tang']['muc_tieu'], 'giam': kb['giam']['muc_tieu']}, 'kich_hoat_truoc': kich_hoat, 'luc_kich_hoat': luc_kh, 'muc_kich_hoat_hom_qua': {'tang': kb['tang']['kich_hoat'], 'giam': kb['giam']['kich_hoat']}, 'gia_that': {'cao': cao, 'thap': thap, 'dong_luc_cat': duong[-1][4]}, 'bien_do_du_kien_hom_qua': ({'tran': bdk['tran_du_kien'], 'san': bdk['san_du_kien'], 'vuot_tran': cao > bdk['tran_du_kien'], 'thung_san': thap < bdk['san_du_kien']} if 'tran_du_kien' in bdk else None), 'vung_da_cham': vung_cham, 'diem_hom_qua': {'trend': sl['diem_trend']['diem'], 'dca': sl['diem_dca']['diem']}, 'tin_da_ra': [{'gio': m['ngay'] + ' ' + m['gio'], 'tin': m['tin'], 'vang_chay_2h': m['phan_ung_that']['bien_do_2h'], 'huong_2h': m['phan_ung_that']['huong_2h']} for m in tin_that], } # --------------------------------------------------------------------- # SÁNG THỨ HAI: TỔNG KẾT PHÂN TÍCH TUẦN # --------------------------------------------------------------------- # 03/10/2026 chủ dự án: "đầu ngày thứ 2 nên là tổng kết phân tích của tuần trước đó". Bản tin thứ Hai = nhìn lại tuần # (nến tuần, từng phiên, xu hướng tuần/ngày, liên thị trường, tin đã ra + phản ứng, tự chấm 5 bản tin) + khung tuần mới # (vùng giá tuần: pivot tuần, đỉnh/đáy tuần, cùng các vùng ngày; lịch tin cả tuần; kịch bản và điểm Trend/DCA đầu tuần). def nen_tuan(d1: List[List[Any]]) -> List[List[Any]]: """Gộp nến D1 thành nến tuần (thứ Hai–thứ Sáu theo ngày sàn): [ngày thứ Hai, o, h, l, c].""" ra: List[List[Any]] = [] for b in d1: d = date.fromisoformat(b[0][:10]) dau = (d - timedelta(days=d.weekday())).isoformat() if ra and ra[-1][0][:10] == dau: ra[-1][2], ra[-1][3], ra[-1][4] = max(ra[-1][2], b[2]), min(ra[-1][3], b[3]), b[4] else: ra.append([dau + ' 00:00', b[1], b[2], b[3], b[4]]) return ra def _gia_truoc(ds: List[List[Any]], moc: str, chi_so: int = 1) -> Optional[Any]: """Phần tử cuối có ngày < moc (ds sắp tăng theo ngày ở cột 0).""" x = [r for r in ds if str(r[0])[:10] < moc] return x[-1][chi_so] if x else None def tong_ket_tuan(goi: Dict[str, Any], ngay: date, gio_cat: int = 7) -> Dict[str, Any]: """Số liệu bài SÁNG CHỦ NHẬT `ngay` (03/10/2026 chủ dự án: "chủ nhật ra bản tin tổng kết cuối tuần", gộp tổng kết + tầm nhìn): tuần giao dịch vừa kết thúc (thứ Hai–thứ Sáu) + khung tuần mới (dựng như sáng thứ Hai sắp tới nhưng chỉ dùng dữ liệu có trước 7:00 sáng Chủ nhật).""" from ban_tin.lich_tin import moc_tin if ngay.weekday() != 6: raise ValueError('Tổng kết tuần chỉ ra sáng Chủ nhật.') thu_hai_moi = ngay + timedelta(days=1) dau = thu_hai_moi - timedelta(days=7) cac_ngay = [dau + timedelta(days=i) for i in range(5)] d1 = [b for b in goi['d1'] if b[0][:10] < thu_hai_moi.isoformat()] tuan = [b for b in d1 if dau.isoformat() <= b[0][:10] < thu_hai_moi.isoformat()] if len(tuan) < 3: raise ValueError('Thiếu nến tuần vừa qua.') wk = nen_tuan(d1) truoc_tuan = [w for w in wk if w[0][:10] < dau.isoformat()] bien_tuan = [w[2] - w[3] for w in truoc_tuan[-10:]] w = wk[-1] dong_truoc = truoc_tuan[-1][4] if truoc_tuan else w[1] reg = {r['date']: r for r in goi.get('regime') or []} # --- từng bản tin trong tuần (dựng lại đúng như sáng hôm đó) + tự chấm của từng bản ban = {} for d in cac_ngay: try: ban[d] = so_lieu_sang(goi, d, gio_cat, tu_cham=False) except ValueError: pass # khung tuần mới = như sáng thứ Hai sắp tới, cắt dữ liệu ở 7:00 sáng Chủ nhật (giờ cắt âm 17 tiếng) hien_tai = so_lieu_sang(goi, thu_hai_moi, gio_cat - 24, tu_cham=False) hien_tai['tan_suat'] = tan_suat_kich_ban(goi, ngay, gio_cat=gio_cat) cham = [] for d in cac_ngay: if d not in ban: continue sau = [x for x in cac_ngay if x > d] try: tc = tu_cham_hom_qua(goi, sau[0] if sau else ngay, gio_cat) except ValueError: continue bdk = tc['bien_do_du_kien_hom_qua'] or {} cham.append({'ngay': d.isoformat(), 'xay_ra': tc['kich_ban_xay_ra'], 'thuan_boi_canh': tc['thuan_boi_canh'], 'trong_bien_do_du_kien': (not bdk.get('vuot_tran') and not bdk.get('thung_san')) if bdk else None, 'so_vung_cham': len(tc['vung_da_cham']), 'diem_trend': ban[d]['diem_trend']['diem'], 'diem_dca': ban[d]['diem_dca']['diem']}) # --- từng phiên phien = [] for i, b in enumerate(tuan): truoc = d1[d1.index(b) - 1][4] r = reg.get(b[0][:10]) phien.append({'ngay': b[0][:10], 'mo': b[1], 'cao': b[2], 'thap': b[3], 'dong': b[4], 'thay_doi': round(b[4] - truoc, 2), 'thay_doi_pct': round((b[4] / truoc - 1) * 100, 2), 'bien_do': round(b[2] - b[3], 2), 'trang_thai': TEN_TRANG_THAI.get(r['regime'], (r['regime'],))[0] if r else None}) # --- xu hướng khung tuần + ngày khung_tuan = khung(wk, 'W1') if len(wk) >= 30 else None # --- liên thị trường cả tuần lien: Dict[str, Any] = {} ux = (goi.get('usdx') or {}).get('d1') or [] u_cuoi, u_dau = _gia_truoc(ux, thu_hai_moi.isoformat(), 4), _gia_truoc(ux, dau.isoformat(), 4) if u_cuoi and u_dau: lien['usdx'] = {'dong_tuan': round(u_cuoi, 3), 'doi_tuan_pct': round((u_cuoi / u_dau - 1) * 100, 2)} ls = goi.get('us10y') or [] y_cuoi, y_dau = _gia_truoc(ls, thu_hai_moi.isoformat()), _gia_truoc(ls, dau.isoformat()) if y_cuoi is not None and y_dau is not None: lien['loi_suat_my_10_nam'] = {'cuoi_tuan_pct': y_cuoi, 'doi_tuan_diem_co_ban': round((y_cuoi - y_dau) * 100)} tq = hien_tai['lien_thi_truong'].get('tuong_quan_vang_usdx_20_phien') if tq is not None: lien['tuong_quan_vang_usdx_20_phien'] = tq # --- tin đã ra trong tuần (giờ VN, thứ Hai 7:00 -> thứ Hai sau 7:00) bd = datetime.combine(dau, datetime.min.time()) + timedelta(hours=7) tin_da_ra = [m for m in moc_tin(goi, bd, bd + timedelta(days=7)) if 'phan_ung_that' in m] # --- khung tuần mới: pivot tuần từ nến tuần vừa qua + vùng ngày của bản thứ Hai p = (w[2] + w[3] + w[4]) / 3 pivot_tuan = {'P': round(p, 2), 'R1': round(2 * p - w[3], 2), 'S1': round(2 * p - w[2], 2), 'R2': round(p + (w[2] - w[3]), 2), 'S2': round(p - (w[2] - w[3]), 2)} return { 'ngay_ban_tin': ngay.isoformat(), 'tuan_tu': dau.isoformat(), 'tuan_den': cac_ngay[-1].isoformat(), 'nen_tuan': {'mo': w[1], 'cao': w[2], 'thap': w[3], 'dong': w[4], 'thay_doi': round(w[4] - dong_truoc, 2), 'thay_doi_pct': round((w[4] / dong_truoc - 1) * 100, 2), 'bien_do': round(w[2] - w[3], 2), 'bien_do_tb_10_tuan': round(sum(bien_tuan) / len(bien_tuan), 2) if bien_tuan else None}, 'phien': phien, 'khung_tuan': khung_tuan, 'khung_ngay': hien_tai['da_khung'][0], 'lien_thi_truong': lien, 'tin_da_ra': tin_da_ra, 'cham': cham, 'pivot_tuan': pivot_tuan, 'gap': hien_tai.get('gap'), 'su_kien_fed': _su_kien(bd, bd + timedelta(days=7)), 'dau_tuan_moi': hien_tai, # vùng giá, lịch tin cả tuần, kịch bản, điểm… dựng cho tuần mới 'd1_ve': d1[-70:], } # --------------------------------------------------------------------- # SÁNG THỨ BẢY: KẾT PHIÊN THỨ SÁU # --------------------------------------------------------------------- # 03/10/2026 chủ dự án chọn: thị trường đóng từ sáng sớm thứ Bảy -> bài ngắn: phiên thứ Sáu, cả tuần đến lúc đóng cửa, # tin đã ra + phản ứng, bối cảnh, hệ thống. KHÔNG ra kịch bản (tới sáng thứ Hai mới mở cửa). def ket_phien_thu_sau(goi: Dict[str, Any], ngay: date, gio_cat: int = 7) -> Dict[str, Any]: if ngay.weekday() != 5: raise ValueError('Kết phiên thứ Sáu chỉ ra sáng thứ Bảy.') sl = so_lieu_sang(goi, ngay, gio_cat) # hom_qua = nến thứ Sáu; tu_cham = chấm bản tin thứ Sáu (báo cáo riêng) tc = sl.get('tu_cham') or {} sl['tin_thu_sau'] = tc.get('tin_da_ra') or [] return sl # --------------------------------------------------------------------- # NGÀY GIAO DỊCH ĐẦU THÁNG: TẦM NHÌN THÁNG # --------------------------------------------------------------------- # 03/10/2026 chủ dự án chọn: bài tầm nhìn tháng ra sáng NGÀY GIAO DỊCH ĐẦU THÁNG (thay bản tin ngày hôm đó). def ngay_dau_thang(ngay: date) -> bool: d = ngay.replace(day=1) while d.weekday() >= 5: d += timedelta(days=1) return ngay == d def tam_nhin_thang(goi: Dict[str, Any], ngay: date, gio_cat: int = 7) -> Dict[str, Any]: from ban_tin.lich_tin import moc_tin d1 = [b for b in goi['d1'] if b[0][:10] < ngay.isoformat()] dau = ngay.replace(day=1) dau_truoc = (dau - timedelta(days=1)).replace(day=1) th = nen_thang([b for b in d1 if b[0][:10] < dau.isoformat()]) if len(th) < 13: raise ValueError('Thiếu nến tháng.') m, m_truoc = th[-1], th[-2] bien_12 = [x[2] - x[3] for x in th[-13:-1]] trong_thang = [b for b in d1 if dau_truoc.isoformat() <= b[0][:10] < dau.isoformat()] tuan_trong = [w for w in nen_tuan(trong_thang)] hien_tai = so_lieu_sang(goi, ngay, gio_cat) # tự chấm cả tháng: mỗi ngày giao dịch của tháng trước cham = [] d = dau_truoc while d < dau: if d.weekday() < 5: sau = d + timedelta(days=1) while sau.weekday() >= 5: sau += timedelta(days=1) try: tc = tu_cham_hom_qua(goi, sau, gio_cat) bdk = tc['bien_do_du_kien_hom_qua'] or {} cham.append({'ngay': d.isoformat(), 'xay_ra': tc['kich_ban_xay_ra'], 'thuan_boi_canh': tc['thuan_boi_canh'], 'trong_bien_do_du_kien': (not bdk.get('vuot_tran') and not bdk.get('thung_san')) if bdk else None}) except (ValueError, IndexError): pass d += timedelta(days=1) lien: Dict[str, Any] = {} ux = (goi.get('usdx') or {}).get('d1') or [] u1, u0 = _gia_truoc(ux, dau.isoformat(), 4), _gia_truoc(ux, dau_truoc.isoformat(), 4) if u1 and u0: lien['usdx'] = {'dong_thang': round(u1, 3), 'doi_thang_pct': round((u1 / u0 - 1) * 100, 2)} ls = goi.get('us10y') or [] y1, y0 = _gia_truoc(ls, dau.isoformat()), _gia_truoc(ls, dau_truoc.isoformat()) if y1 is not None and y0 is not None: lien['loi_suat_my_10_nam'] = {'cuoi_thang_pct': y1, 'doi_thang_diem_co_ban': round((y1 - y0) * 100)} bd_truoc = datetime.combine(dau_truoc, datetime.min.time()) + timedelta(hours=7) bd = datetime.combine(dau, datetime.min.time()) + timedelta(hours=7) tin_qua = [x for x in moc_tin(goi, bd_truoc, bd) if x['hang'] == 1 and 'phan_ung_that' in x] tin_toi = [x for x in moc_tin(goi, datetime.combine(ngay, datetime.min.time()) + timedelta(hours=gio_cat), datetime.combine((dau + timedelta(days=32)).replace(day=1), datetime.min.time())) if x['hang'] == 1] lich_den = max((x['t_utc'] for x in goi.get('lich') or []), default=None) p = (m[2] + m[3] + m[4]) / 3 pivot = {'P': round(p, 2), 'R1': round(2 * p - m[3], 2), 'S1': round(2 * p - m[2], 2), 'R2': round(p + (m[2] - m[3]), 2), 'S2': round(p - (m[2] - m[3]), 2)} return { 'ngay_ban_tin': ngay.isoformat(), 'thang_truoc': dau_truoc.strftime('%m/%Y'), 'thang_moi': dau.strftime('%m/%Y'), 'nen_thang': {'mo': m[1], 'cao': m[2], 'thap': m[3], 'dong': m[4], 'thay_doi': round(m[4] - m_truoc[4], 2), 'thay_doi_pct': round((m[4] / m_truoc[4] - 1) * 100, 2), 'bien_do': round(m[2] - m[3], 2), 'bien_do_tb_12_thang': round(sum(bien_12) / len(bien_12), 2)}, 'tung_tuan': [{'tuan_tu': w[0][:10], 'mo': w[1], 'cao': w[2], 'thap': w[3], 'dong': w[4]} for w in tuan_trong], 'khung_thang': khung(th, 'MN') if len(th) >= 20 else None, 'khung_tuan': khung(nen_tuan(d1), 'W1'), 'boi_canh': hien_tai['boi_canh'], 'lien_thi_truong': lien, 'tin_thang_qua': tin_qua, 'su_kien_fed_thang_qua': _su_kien(bd_truoc, bd), 'cham': cham, 'pivot_thang': pivot, 'tin_thang_moi': tin_toi, 'lich_co_den': (datetime.strptime(lich_den, '%Y-%m-%d %H:%M:%S') + timedelta(hours=7)).strftime('%d/%m') if lich_den else None, 'dau_thang_moi': hien_tai, 'w1_ve': nen_tuan(d1)[-60:], } if __name__ == '__main__': assert lech_san_utc(datetime(2026, 10, 2, 12)) == 3 and lech_san_utc(datetime(2026, 11, 2, 12)) == 2 assert lech_san_utc(datetime(2026, 3, 29, 0, 30)) == 2 and lech_san_utc(datetime(2026, 3, 29, 1, 30)) == 3 assert san_sang_vn('2026-10-02 13:00') == datetime(2026, 10, 2, 17, 0) # hè: sàn +3 -> VN +7 assert san_sang_vn('2026-12-02 13:00') == datetime(2026, 12, 2, 18, 0) # đông: sàn +2 assert moc_tron(4177.3) == {'duoi': 4150, 'tren': 4200} and moc_tron(4200) == {'duoi': 4200, 'tren': 4250} r = rsi([float(x) for x in range(1, 30)]) assert r[-1] == 100.0 and r[5] is None assert abs(ema([1.0] * 10, 3)[-1] - 1.0) < 1e-9 assert ngay_giao_dich_truoc(date(2026, 10, 5)) == date(2026, 10, 2) # xu hướng tăng đều: ADX cao, +DI > -DI, cấu trúc không có sóng nen = [['', 100 + i, 101 + i, 99.5 + i, 100.8 + i] for i in range(80)] a = adx_di(nen) assert a[-1][0] > 50 and a[-1][1] > a[-1][2], a[-1] v = gom_vung([{'ten': 'A', 'gia': 100, 'trong_so': 2}, {'ten': 'B', 'gia': 100.2, 'trong_so': 1.5}, {'ten': 'C', 'gia': 110, 'trong_so': 1}, {'ten': 'D', 'gia': 95, 'trong_so': 3}], 4, 104) assert v['ho_tro'][0]['gom'] == ['A', 'B'] and v['ho_tro'][0]['do_manh'] == 'vừa' and v['khang_cu'][0]['giua'] == 110, v w = nen_tuan([['2026-09-28 00:00', 1, 5, 0.5, 2], ['2026-09-29 00:00', 2, 6, 1, 3], ['2026-10-05 00:00', 3, 4, 2, 3.5]]) assert w == [['2026-09-28 00:00', 1, 6, 0.5, 3], ['2026-10-05 00:00', 3, 4, 2, 3.5]], w assert ngay_dau_thang(date(2026, 10, 1)) and ngay_dau_thang(date(2026, 11, 2)) and not ngay_dau_thang(date(2026, 11, 3)) assert nen_thang([['2026-09-29 00:00', 1, 5, 0.5, 2], ['2026-09-30 00:00', 2, 6, 1, 3], ['2026-10-01 00:00', 3, 4, 2, 3.5]])[0][2:] == [6, 0.5, 3] print('TỰ KIỂM ban_tin/phan_tich.py QUA.')