标签：#Python

Pytest Fundamentals for Data Engineers Data engineering code is still software, and software that doesn't have tests is a liability. If you've ever had a production pipeline silently produce wrong numbers because a downstream transformation made an assumption about a column that changed upstream, you know the pain. Testing isn't just about "does the pipeline run" — it's about verifying that every transformation, every branch of logic, and every edge case behaves exactly as you expect. This post lays the groundwork for testing data pipelines with pytest. I'll cover why testing matters, the fundamental pattern every test follows, how to write your first tests and assertions, how to verify that your code raises the right exceptions, and how to mock external dependencies so your tests stay fast and deterministic. By the end, you'll have a solid foundation for writing unit tests that catch regressions before they reach production. Testing 101 for Data Pipelines If you're new to testing or have only ever written a few scripts, the core idea is simple: a test is a small piece of code that exercises another piece of code and checks that the result is what you expected. In pytest, that looks like a function whose name starts with test_ and which contains an assert statement. The Fundamental Pattern: Arrange, Act, Assert Almost every test follows three steps: Arrange – set up the input data and any necessary context. Act – call the function or method you want to test. Assert – verify the output is correct. For a data pipeline transformation, that might be: def test_clean_amount (): # Arrange raw_data = " 1,200.50 " # what you might get from a CSV # Act result = clean_amount ( raw_data ) # Assert assert result == 1200.50 clean_amount is a pure function — no file I/O, no database calls. It just takes an input string and returns a number. That makes it straightforward to test. Assert Raises: Expecting Exceptions Data pipelines often deal with invalid data that should trigger an

Introduction Fintech engineers building backtesting engines, live quote dashboards, and algorithmic trading pipelines repeatedly face consistent pain points with market data APIs: limited granularity on free tiers, disjoint real-time and historical endpoints, inconsistent protocol support, and fragmented cross-asset coverage. This neutral technical breakdown compares three widely adopted market data providers, evaluating native functionality and end-to-end integration patterns to streamline API vendor selection for backend and quant teams. Core Evaluation Criteria Data Granularity & Historical Depth: Support for tick, intraday, and daily bars plus long-term archived records across equities and FX Protocol Compatibility: Native REST batch query and WebSocket real-time streaming implementation Developer Operational Overhead: Rate limits, documentation completeness, and production integration complexity Comparative Overview Provider Value Propositions AllTick: All-in-one multi-asset market data API built for quant developers, delivering unified tick/intraday/daily historical archives and dual REST/WebSocket access with balanced pricing for individual builders and small teams. Bloomberg: Institutional-grade terminal API offering comprehensive cross-market depth, alternative datasets, and proprietary analytics; targeted exclusively at enterprise investment teams with high entry integration overhead and subscription costs. Alpha Vantage: Lightweight free-first REST API ideal for early-stage prototyping and educational use, lacking native real-time streaming and deep tick-level historical archives. Feature Comparison Matrix Metric AllTick Bloomberg Alpha Vantage Free Tier Rate Limits 100 requests/min, full tick granularity access No permanent free tier; limited trial enterprise access only 5 requests/min, restricted to daily/intraday bars Live Latency Average 170ms native WebSocket push Sub-10ms dedicated institutional line feeds Polling-only, minute-scale delayed refresh

The Problem: Why Most Telegram Payment Bots Fail You've built a Telegram bot that sells digital courses, takes coffee orders, or offers freelance services. Traffic is flowing. But when your bot tries to collect payment, you hit a wall: Payment gateways demand hefty setup fees and compliance audits You end up holding customer money in an intermediary account (legal liability) Settlement takes 3–5 days, frustrating both you and buyers Integration is a maze of webhooks, IPNs, and error handling A founder we know spent two weeks wrestling with Stripe's Telegram integration, only to discover the monthly fee ate 40% of his margins on $2–5 transactions. He needed something lighter, faster, and truly peer-to-peer. Enter AgentPay VN : an open-source Python SDK that lets your Telegram bot generate QR codes pointing directly to your bank account. No middleman. No holding funds. Just instant VietQR payments via the banking system Vietnameses already use daily. What Is AgentPay VN? AgentPay VN is an MIT-licensed Python SDK + MCP server that orchestrates VietQR payment requests. Here's the mental model: You create a payment request in your bot (e.g., "Customer ordered coffee for 50,000 VND") AgentPay generates a checkout URL with an embedded QR code Customer scans → pays → bank confirms settlement in seconds Your bot receives a webhook callback and fulfills the order The key: AgentPay never touches money . The QR points straight at your merchant bank account. A bank feed confirms settlement. You own the transaction end-to-end. Why Telegram + VietQR? Telegram has 180+ million users , with especially strong adoption in Southeast Asia. Vietnamese merchants already use banking apps (MB Bank, Techcombank, VCB) that natively support VietQR scanning. Your bot becomes a natural extension of their daily workflow: Customer receives a payment link in chat Opens the QR code (in-app or screenshot) Scans with their banking app (2–3 taps) Money clears to your account in minutes Bot auto-confirm

It is been while I am learning and build around FastAPI. So there is a project where I was thinking how to add this new feature over exiting one. Like what changes I need to make in database which need to be reflected in my backend and frontend. I already lunched the web locally. Problem started When I when back to the web and reload it it shows this error: ERROR: ConnectTimeout: Unauthorized 401. I was like what? Why? I thougth there is some issue with login endpoint or refresh token function. When i did some debugging and found some new information which is: "Either Supabase's edge/pooler (or OS, or an intermediate proxy/NAT) silently kills those idle connections server-side after some timeout but client-side pool doesn't know that." As I was doing nothing in become idle state so to save the resources server side silently close that particular connection. So I came back and try to connect it give this error. First thought come it my mind after this was there should be a way to automatically check this idle state and if user was in ideal state then create a new connection. Proposed Solutions After a while I come up with these solution: Calculate the Idle time: if it is more then server connection timeout then establish new connection. Retry logic: retry once on the specific connection errors. I thought this will work but This again give me error then this new issue I faced. Cold-start connection problem There is something call dual-stack (IPv4 and IPv6) networks and Happy Eyeballs is a network mechanism which automatically move to IPv4 connection if IPv6 fails. But supabase-py uses httpx and it doesn't support Happy Eyeballs. So in first try after the connection time out it try to establish IPv6 connection which is not routeable in most Pakistani ISPs and ultimately it fails and wait for timeout. There is no way to try it again for IPv4. So we have to do it manually. So this error help me to learn many thing in process. Share your thoughts.

Most "how to add SAST to your pipeline" articles gravitate toward the same four names: SonarQube, Snyk, Semgrep, Veracode. They're solid tools, but they're not the only options, and sometimes you can't use them — budget constraints, air-gapped environments, licensing restrictions, or simply wanting something lightweight that lives entirely in your repo. The OWASP Source Code Analysis Tools page lists dozens of alternatives across every language. In this article I'll walk through applying Bandit , a free, open-source SAST tool for Python, to a real sample application — from finding vulnerabilities locally to wiring it into a CI/CD pipeline with GitHub Actions. The same workflow (install → configure → scan → fail the build on high-severity findings → track results over time) applies almost identically if you swap Bandit for other OWASP-listed tools like Brakeman (Ruby), FindSecBugs (Java), Gosec (Go), or Horusec (multi-language). Why Bandit? 100% open source (Apache 2.0), maintained under the PyCQA org. No account, no server, no license key — it runs as a CLI or a library. Understands Python's AST, so it catches real logic patterns, not just regex matches. Easy to tune with a config file and inline # nosec suppressions. ## 1. The sample application Let's use a small Flask app with a few intentionally introduced vulnerabilities — the kind of thing that slips into real codebases under deadline pressure. # app.py import subprocess import sqlite3 import pickle import yaml from flask import Flask , request app = Flask ( __name__ ) DB_PATH = " users.db " @app.route ( " /ping " ) def ping (): host = request . args . get ( " host " ) # Vulnerable: command injection via shell=True result = subprocess . run ( f " ping -c 1 { host } " , shell = True , capture_output = True ) return result . stdout @app.route ( " /user " ) def get_user (): user_id = request . args . get ( " id " ) conn = sqlite3 . connect ( DB_PATH ) cursor = conn . cursor () # Vulnerable: SQL injection via strin

GitHub热门项目 | AI coding assistant skill (Claude Code, Codex, OpenCode, Cursor, Gemini CLI, and more). Turn any folder of code, SQL schemas, R scripts, shell scripts, docs, papers, images, or videos into a queryable knowledge graph. App code + database schema + infrastructure in one graph. | Stars: 80,027 | 885 stars today | 语言: Python

GitHub热门项目 | | Stars: 2,604 | 14 stars today | 语言: Python

GitHub热门项目 | SkillOpt is a text-space optimizer that trains reusable natural-language skills for frozen LLM agents through trajectory-driven edits, validation-gated updates, and deployable best_skill.md artifacts. | Stars: 11,571 | 261 stars today | 语言: Python

GitHub热门项目 | 🧠「大模型」2小时完全从0训练64M的小参数LLM！Train a 64M-parameter LLM from scratch in just 2h! | Stars: 52,981 | 176 stars today | 语言: Python

Cryptographic Watermarking for LLM Outputs with resk-mark Links: PyPI: https://pypi.org/project/reskmark GitHub: https://github.com/Resk-Security/resk-mark Web: https://resk.fr __ __ ________ _____/ /__ ____ ___ ____ ______/ /__ / ___/ _ \/ ___/ //_/_____/ __ `__ \/ __ `/ ___/ //_/ / / / __(__ ) ,< /_____/ / / / / / /_/ / / / ,< /_/ \___/____/_/|_| /_/ /_/ /_/\__,_/_/ /_/|_| The Provenance Problem Every company deploying LLMs in production faces the same question: once a model generates text, how do you prove it came from your system? Prompts like "say you are an AI" are trivially removable. Post-hoc detectors are unreliable and adversarial. And once text leaves your system — forwarded, copied, pasted into a ticket — you have zero visibility. resk-mark solves this by embedding a cryptographic watermark directly into the token generation process. The output reads naturally, but carries a verifiable signature that survives rewording and truncation. How It Works resk-mark hooks into the language model's sampling process. Before generation, the caller provides a secret key. During sampling, the library biases the probability distribution toward tokens that encode that key's signature: from reskmark import WatermarkEncoder , verify encoder = WatermarkEncoder ( secret_key = " your-key-here " ) model = AutoModelForCausalLM . from_pretrained ( " mistralai/Mistral-7B " ) # Wrap the generate call output = encoder . generate ( model , " Explain the concept of zero-knowledge proofs. " , max_length = 200 , ) print ( output ) # "Zero-knowledge proofs are a cryptographic method where..." # Reads naturally - watermark is invisible # Later - verify provenance is_authentic , confidence = verify ( output , public_key = " corresponding-pub-key " ) print ( f " Authentic: { is_authentic } , confidence: { confidence : . 2 f } " ) Key Properties Invisible — the watermark does not change the meaning, grammar, or fluency of the output Robust — survives copy, paste, truncation, and light rewo

TL;DR We're building a script that takes a video in English and produces the same video narrated in Spanish, in a cloned version of the original speaker's voice. Stack: faster-whisper for timestamped transcription, an LLM (or any MT engine) for translation, XTTS-v2 for voice-cloned synthesis, FFmpeg for surgery. We'll also handle the problem every demo skips: translated audio that doesn't fit its time slot. 📦 Code: github.com/USER/repo (replace before publishing) If you'd rather start from a finished system, Softcatala's open-dubbing and KrillinAI are full pipelines behind one CLI. This post builds the minimal version by hand so you understand what those tools are doing, and where they break. 0. Setup and a licensing warning ⚠️ Python 3.10–3.12. The original Coqui company shut down in early 2024; the maintained fork of their TTS library is published by Idiap as coqui-tts : $ python -m venv dub && source dub/bin/activate $ pip install faster-whisper coqui-tts $ ffmpeg -version | head -1 # 6.0+ is fine, 8.x current ⚠️ Note: the XTTS-v2 model weights ship under the Coqui Public Model License, which restricts commercial use. Prototype freely, but before dubbed videos ship to paying customers, someone must read that license and possibly swap the synthesis step for a commercially licensed model or paid API. Voice cloning also requires the speaker's consent. Get it in writing. 1. Extract audio and transcribe with word timestamps 🎙️ # pull mono 16k audio for the ASR step $ ffmpeg -i input.mp4 -vn -ac 1 -ar 16000 -y source.wav # dub/transcribe.py from faster_whisper import WhisperModel model = WhisperModel ( " large-v3-turbo " , compute_type = " int8 " ) segments , info = model . transcribe ( " source.wav " , word_timestamps = True ) lines = [] for seg in segments : lines . append ({ " start " : seg . start , " end " : seg . end , " text " : seg . text . strip (), }) print ( f " language= { info . language } segments= { len ( lines ) } " ) The timestamps are the skeleton of

一个任务的奇幻漂流：同一个Agent任务，在FROST和FROST-SOP中分别长什么样？ 作者 ：神通说 日期 ：2026-07-08 主题 ：双项目联动 | 周三轮换 阅读时间 ：8分钟 一个问题 你可能听过这样的故事： "这个框架很好，但我不知道怎么用到真实项目里。" FROST 社区里也收到过类似的反馈—— "500行代码确实让我理解了Agent的本质，但理解完之后呢？怎么从'看懂'到'会用'？" 今天这篇文章，就用 同一个真实任务 ，分别展示它在 FROST 和 FROST-SOP 中的样子。你会发现：它们不是两个不同的东西，而是 同一个东西的不同分辨率 。 今天的任务：自动写日报 假设你是一个独立开发者，想让Agent每天自动帮你写工作日报。需求很简单： 收集今天完成的任务 用LLM生成日报摘要 发送邮件给自己 就这么三步。让我们看看它在两个项目中分别怎么实现。 第一站：FROST——用最少的代码理解本质 FROST 的哲学是： 先用500行代码告诉你Agent的底层逻辑，剩下的你自然就会了。 在FROST中，一个Agent的运作只需要四个原子： from core import Store , Agent , skill_set , skill_get # 1. 创建记忆容器 store = Store () # 2. 定义能力（Skill = 纯函数） def collect_tasks ( context ): """ 收集今日任务 """ tasks = [ " 完成FROST v5.0文档 " , " 修复Skill测试用例 " , " 写推广文章 " ] context [ " tasks " ] = tasks return context def generate_summary ( context ): """ 生成日报摘要（简化版，实际调用LLM） """ tasks = context . get ( " tasks " , []) context [ " summary " ] = f " 今日完成 { len ( tasks ) } 项任务： " + " 、 " . join ( tasks ) return context def send_report ( context ): """ 发送日报 """ context [ " sent " ] = True print ( f " [日报已发送] { context [ ' summary ' ] } " ) return context # 3. 组装Agent agent = Agent ( " daily_reporter " , store , skills = { " collect " : collect_tasks , " summarize " : generate_summary , " send " : send_report }) # 4. 用SOP编排执行顺序 result = agent . run ( sop_steps = [ " collect " , " summarize " , " send " ], initial_context = {} ) # 输出：[日报已发送] 今日完成3项任务：完成FROST v5.0文档、修复Skill测试用例、写推广文章 这段代码做了什么？ Store 是记忆——Agent的工作空间，所有中间结果存在这里 Agent 是细胞——拥有记忆和能力的最小执行单元 sop_steps 是宪法——定义执行顺序，Agent不会自作主张改变流程 就这么简单。没有配置文件，没有YAML，没有复杂的初始化。 30行Python代码，一个完整的Agent就跑起来了。 这就是FROST的价值——它不帮你"做"什么，它帮你 看懂 Agent到底是什么。 问题来了 上面的代码能跑，但如果你真的想把它用在生产环境，你会遇到一堆问题： 发邮件的能力怎么写？ FROST不管——它只告诉你Skill是纯函数，具体实现你自己来 LLM调用怎么复用？ 每次写日报都要调LLM，总不能每次都重写一遍 任务失败了怎么办？ 发邮件失败了，要不要重试？重试几次？ 执行日志在哪？ 老板问"今天日报发了吗"，你怎么知道发没发？ 多个Agent协作呢？ 一个人有好几个项目，每个项目一个Agent，怎么协调？ FROST对这些问题的回答是： 这些问题不是我该回答的。 但它的兄弟——FROST-SOP——就是专门回答这些问题的。 第二站：FROST-SOP——让同一个任务跑在生产环境 FROST-SOP 的哲学是： 把FROST教你的每一个概念，都变成生产级的工程组件。 同样的"自动写日报"任务，在FROST-SOP中长

We’ve all been there: staring at a stack of printed lab results or a folder full of cryptic report_final_v2_NEW.pdf files, trying to remember if our cholesterol was higher or lower two years ago. For developers, this isn't just a filing problem—it's a data engineering challenge. In the world of healthcare, data is messy, siloed, and often locked in "unstructured" formats. To build a truly personal Electronic Health Record (EHR) system, we need more than just a folder; we need a RAG (Retrieval-Augmented Generation) pipeline that can parse PDFs, map them to the FHIR (Fast Healthcare Interoperability Resources) standard, and provide natural language insights. In this guide, we’ll leverage Unstructured.io , Milvus , and DuckDB to turn chaotic medical PDFs into a queryable, structured knowledge base. The Architecture: From Raw Pixels to Structured Insights Before we dive into the code, let’s look at how the data flows from a messy lab report to a structured answer. graph TD A[Unstructured PDF Reports] --> B[Unstructured.io Partitioning] B --> C{Data Split} C -->|Textual Context| D[Milvus Vector DB] C -->|Tabular Data| E[DuckDB Structured Storage] D --> F[LangChain RAG Engine] E --> F G[User Query: Is my glucose trending up?] --> F F --> H[FHIR-Formatted Response] Why this stack? Unstructured.io : The gold standard for handling "ugly" PDFs (tables, headers, and nested lists). Milvus : A high-performance vector database built for scale. DuckDB : Perfect for running complex analytical SQL queries on the extracted "structured" parts of our medical data. FHIR Standard : To ensure our data follows global healthcare interoperability rules. Prerequisites Make sure you have your environment ready: pip install langchain milvus unstructured[pdf] duckdb openai Step 1: Extraction with Unstructured.io Medical PDFs often contain complex tables. Standard PDF parsers usually fail here. We’ll use unstructured to partition the document into logical elements. from unstructured.partition.pdf

GitHub热门项目 | An OSINT tool to search for accounts by username and email in social networks. | Stars: 6,806 | 53 stars today | 语言: Python

GitHub热门项目 | An AI-boost cross-platform multi-protocol fluent-design concurrent downloader built with Python & Qt. | Stars: 5,688 | 59 stars today | 语言: Python

GitHub热门项目 | Production-grade platform for building agentic IM bots - 生产级多平台智能机器人开发平台/ Agent、知识库编排、插件系统 / Bots for Discord / Slack / LINE / Telegram / WeChat(企业微信, 企微智能机器人, 公众号) / 飞书 / 钉钉 / QQ / Matrix e.g. Integrated with ChatGPT(GPT), DeepSeek, Dify, n8n, Langflow, Coze, Claude, Gemini, GLM, Ollama, SiliconFlow, Moonshot, openclaw / hermes agent, deerflow | Stars: 16,733 | 33 stars today | 语言: Python

GitHub热门项目 | A TTS that fits in your CPU (and pocket) | Stars: 5,746 | 510 stars today | 语言: Python

Most of the "AI for observability" work I see right now hands the language model the judgment. I think that's backwards. Feed it the alert, feed it some metrics, ask it what's wrong, what should be done, and let it make the judgement call. Based on my experience working with language models, I decided that inverting the process provides better results. The short version: in my alerting pipeline, the set of allowable classifications is fixed in deterministic Python, and the model has to pick from it. The LLM's only job is to turn a structured verdict into an easily digestible sentence. It never decides whether something is bad, how bad it is, or what category of problem it is. It narrates within a decision space the code has already locked down. TL;DR: Instead of letting an LLM decide what's wrong with an alert, I let deterministic Python make every operational decision and restrict the model to explaining the result in plain English. The code classifies, validates, and aggregates; the LLM only narrates. That keeps the data consistent, prevents hallucinated classifications, and ensures the monitoring pipeline continues working even if the model fails. The problem I run a small managed monitoring service. Alertmanager fires, a webhook lands, and historically that webhook produced a line like HighMemoryUsage on host web-vm, severity warning, which is accurate, but not terribly helpful. The person reading it still has to know what HighMemoryUsage implies, whether this host always runs hot, and whether to care. I wanted plain-English context attached to the alerts without altering the alert delivery process. The obvious move was to throw the whole alert at an LLM and ask it to explain. I tried that in the first iteration of this experiment, expecting it to be somewhat accurate, but not entirely reliable, and it did not disappoint, the model was confidently inconsistent. The same alert, fired three times, produced three different "root cause" categories. One run called a

Most backtests lie to you. Not intentionally. But they lie. You design a strategy, run it on historical data, and watch the returns look incredible. Then you run it live — and it underperforms a simple buy-and-hold from day one. The math wasn't wrong. The data was. If you're: testing momentum or mean-reversion strategies in Python, building quant tools for personal or professional use, or tired of backtests that collapse the moment real execution begins, This changes how you work. TL;DR What this covers: Backtesting trading strategies in Python using EODHD's historical OHLCV data API Stack: requests , pandas , numpy — no heavy frameworks (no backtrader, no vectorbt) Scripts included: Script 1 — Fetch adjusted historical price data from EODHD Script 2 — SMA crossover strategy (20/50-day) Script 3 — RSI mean-reversion strategy Script 4 — Performance metrics: Sharpe ratio, max drawdown, win rate EODHD pricing: Free tier available; full access from $19.99/month Best for: Developers and analysts who need reliable, split/dividend-adjusted data without scraping The Problem with Free Data Most developers start with Yahoo Finance or a scraped CSV. That works fine for a quick prototype. It stops working the moment your strategy includes anything that happened around a stock split, dividend payment, or ticker change. Non-adjusted price data creates ghost signals. A stock "drops 50%" when it actually split 2:1. Your moving average calculates a crossover that never happened in real life. Your strategy looks profitable because it's trading on a data artifact. The free path costs you accuracy. And in backtesting, accuracy is the whole point. The Fix Is Simpler Than You Think The real bottleneck isn't the strategy logic. It's the data source. Use split- and dividend-adjusted closing prices from a reliable provider, and half your backtest reliability problems disappear before you write a single signal. EODHD APIs provides exactly this. Their historical data endpoint returns adjusted

How I Built 7 Apify Actors and Started Earning Passive Income from Web Scraping A few weeks ago I had zero Apify actors. Now I have seven, all published on the Apify Store, monetized with pay-per-event pricing, and slowly building a passive income stream. Here's exactly how I did it — the strategy, the tech stack, the mistakes, and what I'd do differently. The Strategy: Zero Competition Most new Apify developers go after hot categories. LinkedIn scrapers, Amazon product extractors, Twitter data. Makes sense — those have demand. But they also have dozens of established actors with hundreds of reviews. I took the opposite approach. Find niches with zero existing actors. This means lower total addressable market, but 100% of whatever traffic exists goes to you. No competing on price, no fighting for reviews, no SEO war against actors with years of history. How I found the niches: Browsed Apify Store categories sorted by actor count Searched for common developer pain points with no existing Apify solution Checked search volume for "[keyword] API" and "[keyword] scraper" Verified zero results on Apify Store for each candidate The winners: domain intelligence, screenshot comparison, Swedish company registry, IP geolocation, QR code generation, and link metadata extraction. The Tech Stack Every actor uses the same foundation. Apify Python SDK v3.4 handles input/output, storage, proxy, and deployment. Playwright for JavaScript-heavy sites and screenshots. aiohttp for lightweight API scraping (way faster than a full browser). Pillow for image processing. Deployment is one command: apify push The Actors {{domain-intel}} WHOIS, DNS, SSL, and tech stack in one API call. Uses socket + ssl + python-whois for data collection, no external API dependency. $0.005 per run. {{screenshot-api}} Full-page screenshots via Playwright. Handles lazy-loading, infinite scroll, and viewport sizing. $0.003 per run. {{metadata-extractor}} Open Graph, Twitter Cards, JSON-LD, and meta tags from any