# -*- coding: utf-8 -*- """KET_QUA_BOT.PY — số liệu bản tin kết quả bot (phần B), đọc CSDL web. Mọi con số do code tính. Khung "hôm qua" = NGÀY GIAO DỊCH CỦA SÀN gần nhất trước ngày ra bản tin (00:00–24:00 giờ sàn). - SMART DCA: KẾT QUẢ PERDAY — bảng dca_ma_ngay (api1 chấm sau mỗi lần nạp perday rồi đẩy lên web; trang /pages/dca_signals gọi là "sổ tiêu chuẩn": 10 mã signal, vốn danh nghĩa $10.000/mã, KHÔNG phải tài khoản thật), ngày = ngày sàn. 03/10/2026 chủ dự án: "chỉ cần hiển thị lãi lỗ là đc, ko cần hiển thị các cột khác, thống kê theo giờ sàn từ 0h-24h ngày hôm đó, kết quả dựa trên perday, ko theo tk thật" -> bỏ hẳn phần tài khoản chạy thật (stream_tk) khỏi bài B. - VipTrend: stream_lenh nguon tin_hieu/tay (giờ epoch thật), lệnh đóng trong ngày sàn, lãi/lỗ theo ĐIỂM GIÁ (chốt 02/10). """ from __future__ import annotations from datetime import date, datetime, timedelta, timezone from typing import Any, Dict, List, Optional try: from congcu import csdl except ImportError: # chạy lẻ import os import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from congcu import csdl # type: ignore def _vn_tu_epoch(t: int) -> datetime: return datetime.fromtimestamp(int(t), tz=timezone.utc).replace(tzinfo=None) + timedelta(hours=7) def _san_tu_epoch(t: int) -> datetime: """Epoch thật -> đồng hồ sàn (EET/EEST) — giờ lệnh ghi theo giờ sàn cho khớp ngày báo cáo 0h–24h giờ sàn.""" from ban_tin.phan_tich import lech_san_utc utc = datetime.fromtimestamp(int(t), tz=timezone.utc).replace(tzinfo=None) return utc + timedelta(hours=lech_san_utc(utc)) def _epoch_vn(t: datetime) -> int: return int((t - timedelta(hours=7)).replace(tzinfo=timezone.utc).timestamp()) def ngay_giao_dich_truoc(ngay: date) -> date: d = ngay - timedelta(days=1) while d.weekday() >= 5: d -= timedelta(days=1) return d def _viptrend(tu: datetime, den: datetime, _lay_ket_noi=None): """VipTrend (stream_lenh nguon tin_hieu/tay, epoch thật): lệnh đóng trong [tu, den) giờ VN và lệnh còn mở lúc den.""" def _doc(ct): # --- VipTrend --- ct.execute('SELECT id, ma, ten_hien_thi, khung, tre_phut FROM stream_kenh WHERE cong_khai = 1 ORDER BY thu_tu, id') kenh = {r['id']: dict(r) for r in ct.fetchall() or []} ct.execute("SELECT kenh_id, type, gia_mo, gia_dong, time_mo, time_dong, comment FROM stream_lenh " "WHERE nguon IN ('tin_hieu','tay') AND ((time_dong >= %s AND time_dong < %s) OR time_dong IS NULL " "OR time_dong >= %s)", (_epoch_vn(tu), _epoch_vn(den), _epoch_vn(den))) dong, mo = [], [] for r in ct.fetchall() or []: k = kenh.get(r['kenh_id']) if not k: continue huong = 'Mua' if int(r['type']) == 0 else 'Bán' if r['time_dong'] is not None and _epoch_vn(tu) <= int(r['time_dong']) < _epoch_vn(den) and r['gia_dong']: diem = (float(r['gia_dong']) - float(r['gia_mo'])) * (1 if huong == 'Mua' else -1) ly_do = str(r['comment'] or '').split('|')[1] if '|' in str(r['comment'] or '') else '' dong.append({'bot': k['ten_hien_thi'], 'ma': k['ma'], 'khung': k['khung'], 'huong': huong, 'vao': float(r['gia_mo']), 'ra': float(r['gia_dong']), 'diem': round(diem, 2), 'ly_do': {'SL': 'chạm SL', 'TP': 'chạm TP'}.get(ly_do.upper(), ly_do or 'đóng theo tín hiệu'), 'gio_vao': _san_tu_epoch(r['time_mo']).strftime('%d/%m %H:%M'), 'gio_ra': _san_tu_epoch(r['time_dong']).strftime('%d/%m %H:%M')}) elif (r['time_dong'] is None or int(r['time_dong']) >= _epoch_vn(den)) and \ int(r['time_mo']) <= _epoch_vn(den) - int(k['tre_phut'] or 0) * 60: mo.append({'bot': k['ten_hien_thi'], 'huong': huong, 'vao': float(r['gia_mo']), 'gio_vao': _san_tu_epoch(r['time_mo']).strftime('%d/%m %H:%M')}) return dong, mo return csdl.voi_ct(_lay_ket_noi, _doc) def so_lieu_bot(ngay: date, _lay_ket_noi=None) -> Dict[str, Any]: ngay_nen = ngay_giao_dich_truoc(ngay) from ban_tin.phan_tich import san_sang_vn tu = san_sang_vn(f'{ngay_nen.isoformat()} 00:00') den = san_sang_vn(f'{(ngay_nen + timedelta(days=1)).isoformat()} 00:00') def _doc(ct): # --- SMART DCA: lãi / lỗ perday của 10 mã --- ct.execute('SELECT ma, loi_nhuan FROM dca_ma_ngay WHERE ngay = %s ORDER BY loi_nhuan DESC', (ngay_nen.isoformat(),)) dca = [{'ma': r['ma'], 'lai': float(r['loi_nhuan'])} for r in ct.fetchall() or []] return dca dca = csdl.voi_ct(_lay_ket_noi, _doc) dong, mo = _viptrend(tu, den, _lay_ket_noi) return { 'ngay_ban_tin': ngay.isoformat(), 'tu': tu.strftime('%d/%m %H:%M'), 'den': den.strftime('%d/%m %H:%M'), 'khung': f"ngày {ngay_nen.strftime('%d/%m/%Y')}, 0h–24h giờ sàn", 'ngay_nen': ngay_nen.isoformat(), 'dca': dca, 'dca_tong': ({'lai': round(sum(x['lai'] for x in dca), 2), 'so_ma_lai': sum(1 for x in dca if x['lai'] > 0), 'so_ma_lo': sum(1 for x in dca if x['lai'] < 0), 'so_ma': len(dca)} if dca else None), 'viptrend_dong': dong, 'viptrend_mo': mo, 'viptrend_tong_diem': round(sum(x['diem'] for x in dong), 2), } def so_lieu_bot_tuan(ngay: date, _lay_ket_noi=None) -> Dict[str, Any]: """Kết quả bot CẢ TUẦN cho sáng Chủ nhật `ngay`: thứ Hai 00:00 -> thứ Bảy 00:00 giờ sàn. SMART DCA = cộng lãi/lỗ perday 5 ngày của từng mã; VipTrend = lệnh đóng trong tuần + lệnh còn mở lúc đóng cửa thứ Sáu.""" from ban_tin.phan_tich import san_sang_vn if ngay.weekday() != 6: raise ValueError('Kết quả bot tuần chỉ ra sáng Chủ nhật.') thu_hai = ngay - timedelta(days=6) thu_sau = thu_hai + timedelta(days=4) tu = san_sang_vn(f'{thu_hai.isoformat()} 00:00') den = san_sang_vn(f'{(thu_sau + timedelta(days=1)).isoformat()} 00:00') ngay_ds = {(thu_hai + timedelta(days=i)).isoformat() for i in range(5)} def _doc(ct): ct.execute('SELECT ma, SUM(loi_nhuan) lai FROM dca_ma_ngay WHERE ngay >= %s AND ngay <= %s GROUP BY ma ORDER BY lai DESC', (thu_hai.isoformat(), thu_sau.isoformat())) return [{'ma': r['ma'], 'lai': round(float(r['lai']), 2)} for r in ct.fetchall() or []] dca = csdl.voi_ct(_lay_ket_noi, _doc) vt = _viptrend(tu, den, _lay_ket_noi) return { 'ngay_ban_tin': ngay.isoformat(), 'ngay_nen': thu_sau.isoformat(), 'tieu_de': f"Kết quả bot tuần {thu_hai.strftime('%d/%m')}–{thu_sau.strftime('%d/%m/%Y')}", 'khung': f"thứ Hai {thu_hai.strftime('%d/%m')} → thứ Sáu {thu_sau.strftime('%d/%m/%Y')}, giờ sàn", 'dca': dca, 'dca_tong': ({'lai': round(sum(x['lai'] for x in dca), 2), 'so_ma_lai': sum(1 for x in dca if x['lai'] > 0), 'so_ma_lo': sum(1 for x in dca if x['lai'] < 0), 'so_ma': len(dca)} if dca else None), 'viptrend_dong': vt[0], 'viptrend_mo': vt[1], 'viptrend_tong_diem': round(sum(x['diem'] for x in vt[0]), 2), '_ngay_ds': sorted(ngay_ds), } def them_tam_tinh(kq: Dict[str, Any], gia_dong: Optional[float]) -> Dict[str, Any]: """Lệnh VipTrend còn mở cuối ngày: lãi/lỗ TẠM TÍNH theo điểm giá tại giá đóng cửa D1 của ngày sàn (chưa chốt). 03/10/2026 chủ dự án: "phần lệnh đang mở của signal cũng báo đi kèm".""" kq['gia_dong_ngay'] = gia_dong for x in kq.get('viptrend_mo') or []: x['tam_tinh'] = (round((gia_dong - x['vao']) * (1 if x['huong'] == 'Mua' else -1), 2) if gia_dong is not None else None) return kq if __name__ == '__main__': assert _vn_tu_epoch(0) == datetime(1970, 1, 1, 7) t = int(datetime(2026, 9, 29, 0, 30, tzinfo=timezone.utc).timestamp()) # 07:30 VN = 03:30 giờ sàn (hè) assert _san_tu_epoch(t) == datetime(2026, 9, 29, 3, 30) assert _epoch_vn(datetime(2026, 10, 2, 7)) == int(datetime(2026, 10, 2, 0, tzinfo=timezone.utc).timestamp()) assert ngay_giao_dich_truoc(date(2026, 10, 5)) == date(2026, 10, 2) and ngay_giao_dich_truoc(date(2026, 10, 2)) == date(2026, 10, 1) k = them_tam_tinh({'viptrend_mo': [{'huong': 'Mua', 'vao': 100.0}, {'huong': 'Bán', 'vao': 100.0}]}, 90.5) assert [x['tam_tinh'] for x in k['viptrend_mo']] == [-9.5, 9.5] print('TỰ KIỂM ban_tin/ket_qua_bot.py QUA.')