Compare commits
8 Commits
bb3d4e8c7f
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d126c5d59d | ||
|
|
8be7f745b1 | ||
|
|
7ce88e6e4a | ||
|
|
65a1485ff9 | ||
|
|
7d9dfa309c | ||
|
|
ed5d126d77 | ||
|
|
51193ab933 | ||
|
|
ee0f05f583 |
168
README.md
168
README.md
@@ -16,7 +16,12 @@ Tento projekt je určen pro stahování a správu kurzů cizích měn vůči če
|
||||
- **Kontrola konzistence ročních dat**: Při startu programu automaticky kontroluje, zda roční data pro aktuální rok obsahují záznamy za poslední 3 pracovní dny. Pokud ne, data jsou automaticky aktualizována.
|
||||
- **Automatické stahování ročních dat**: Pokud jsou požadována data pro rok, který v databázi není, program automaticky stáhne roční data pro daný rok, aktualizuje databázi a teprve poté vrátí výsledek.
|
||||
- **Generování reportů**: Lze vygenerovat report kurzů pro zadaný rok, měsíc nebo časové období včetně dopočítaných kurzů pro dny, kdy ve vstupních datech neexistovali.
|
||||
- **Správné dopočítání kurzů**: Program správně aplikuje pravidla ČNB pro dopočítání kurzů pro víkendy a svátky jak při vyhledávání (`--get-rate`), tak při generování reportů.
|
||||
- **Správné dopočítání kurzů**: Program správně aplikuje pravidla ČNB pro dopočítání kurzů pro víkendy a svátky jak při vyhledávání (`--get-rate`), tak při generování reportů.
|
||||
- **Výpočet Jednotného kurzu**: Lze vypočítat 'Jednotný kurz' pro daňové účely podle metodiky ČNB jako aritmetický průměr kurzů k posledním dnům každého měsíce v roce.
|
||||
- **Validace dat**: Program umí validovat data pro konzistenci, detekovat změny kurzů přesahující prahové hodnoty, kontrolovat počet obchodních dnů a analyzovat časové mezery v datech.
|
||||
- **Analýza počtu záznamů**: Lze zobrazit počty záznamů podle různých časových období (týden, měsíc, čtvrtletí, pololetí, rok).
|
||||
- **Adaptivní prahy**: Systém se učí z historických dat a automaticky upravuje prahy pro detekci anomálií.
|
||||
- **JSON výstup**: Všechny příkazy podporují JSON formát pro programové zpracování pomocí přepínače `--json`.
|
||||
|
||||
## Požadavky
|
||||
|
||||
@@ -60,8 +65,15 @@ Při každém spuštění programu:
|
||||
- `--date DATUM`: Stáhne denní data pro zadané datum. Formát: `DD.MM.YYYY`.
|
||||
- `--get-rate DATUM`: Vyhledá kurz pro zadané datum. Formát: `DD.MM.YYYY`. Vyžaduje `-c` nebo `--currency`.
|
||||
- `--auto-download`: Povolí automatické stahování denních dat pro dnešní datum, pokud je po 14:30 a kurz není k dispozici.
|
||||
- `--report-year ROK [--report-month MESIC]`: Vygeneruje report kurzů pro zadaný rok (a případně měsíc). Vyžaduje `-c` nebo `--currency`.
|
||||
- `--report-period ZACATEK KONEC`: Vygeneruje report kurzů pro zadané časové období. Vyžaduje `-c` nebo `--currency`.
|
||||
- `--report-year ROK [--report-month MESIC]`: Vygeneruje report kurzů pro zadaný rok (a případně měsíc). Vyžaduje `-c` nebo `--currency`.
|
||||
- `--report-period ZACATEK KONEC`: Vygeneruje report kurzů pro zadané časové období. Vyžaduje `-c` nebo `--currency`.
|
||||
- `--stats [ROK]`: Vypočítá 'Jednotný kurz' pro daňové účely podle metodiky ČNB. Pokud je zadán rok, vytvoří kurz pro konkrétní rok. Pokud není rok zadán, vytvoří kurzy pro všechny roky s dostupnými daty. Vyžaduje `-c` nebo `--currency`.
|
||||
- `--validate`: Validuje data pro měnu nebo všechny měny. Zkontroluje konzistenci kurzů, počet obchodních dnů a detekuje možné chyby.
|
||||
- `--record-counts`: Zobrazí počet záznamů podle časových období (týden, měsíc, čtvrtletí, pololetí, rok). Vyžaduje `-c` nebo `--currency`.
|
||||
- `--change-threshold PRAH`: Práh pro detekci změn kurzů v procentech (výchozí: 1.0).
|
||||
- `--no-adaptive`: Vypne adaptivní učení prahů na základě historických dat.
|
||||
- `--gap-threshold DNY`: Maximální přijatelná mezera v pracovních dnech (výchozí: 3).
|
||||
- `--json`: Výstup ve formátu JSON místo prostého textu pro programové zpracování.
|
||||
|
||||
### Příklady
|
||||
|
||||
@@ -111,6 +123,156 @@ Při každém spuštění programu:
|
||||
python src/cli.py --report-period 01.07.2020 31.07.2020 -c USD
|
||||
```
|
||||
|
||||
10. **Výpočet Jednotného kurzu pro daňové účely pro USD za rok 2025**:
|
||||
```bash
|
||||
python src/cli.py --stats 2025 -c USD
|
||||
```
|
||||
|
||||
11. **Výpočet Jednotného kurzu pro daňové účely pro všechny roky s daty**:
|
||||
```bash
|
||||
python src/cli.py --stats -c USD
|
||||
```
|
||||
|
||||
12. **Validace dat pro měnu USD za rok 2025**:
|
||||
```bash
|
||||
python src/cli.py --validate --currency USD --year 2025
|
||||
```
|
||||
|
||||
13. **Validace všech měn s vlastními prahy**:
|
||||
```bash
|
||||
python src/cli.py --validate --change-threshold 0.5 --gap-threshold 2
|
||||
```
|
||||
|
||||
14. **Zobrazení počtu záznamů podle časových období pro USD**:
|
||||
```bash
|
||||
python src/cli.py --record-counts --currency USD --year 2025
|
||||
```
|
||||
|
||||
15. **Získání posledního dostupného kurzu USD**:
|
||||
```bash
|
||||
python src/cli.py -c USD
|
||||
```
|
||||
|
||||
16. **JSON výstup pro vyhledání kurzu**:
|
||||
```bash
|
||||
python src/cli.py --get-rate 01.01.2025 -c USD --json
|
||||
```
|
||||
|
||||
17. **JSON výstup pro validaci dat**:
|
||||
```bash
|
||||
python src/cli.py --validate --currency USD --year 2025 --json
|
||||
```
|
||||
|
||||
## JSON formát
|
||||
|
||||
Při použití přepínače `--json` program vrací strukturovaná data ve formátu JSON:
|
||||
|
||||
### Jednotlivý kurz
|
||||
```json
|
||||
{
|
||||
"currency": "USD",
|
||||
"rate": 24.214,
|
||||
"date": "01.01.2025",
|
||||
"timestamp": "2025-01-12T10:30:00Z"
|
||||
}
|
||||
```
|
||||
|
||||
### Jednotný kurz pro jeden rok
|
||||
```json
|
||||
{
|
||||
"currency": "USD",
|
||||
"year": 2025,
|
||||
"unified_rate": 21.84,
|
||||
"calculation_date": "2025-01-12T10:30:00Z"
|
||||
}
|
||||
```
|
||||
|
||||
### Jednotný kurz pro více let
|
||||
```json
|
||||
{
|
||||
"currency": "USD",
|
||||
"results": [
|
||||
{"year": 2023, "unified_rate": 23.45},
|
||||
{"year": 2024, "unified_rate": 23.28},
|
||||
{"year": 2025, "unified_rate": 21.84}
|
||||
],
|
||||
"calculation_date": "2025-01-12T10:30:00Z"
|
||||
}
|
||||
```
|
||||
|
||||
### Poslední dostupný kurz
|
||||
```json
|
||||
{
|
||||
"currency": "USD",
|
||||
"rate": 20.632,
|
||||
"date": "31.12.2025",
|
||||
"timestamp": "2025-01-12T10:30:00Z"
|
||||
}
|
||||
```
|
||||
|
||||
### Chyba
|
||||
```json
|
||||
{
|
||||
"error": "Kurz nebyl nalezen",
|
||||
"code": "RATE_NOT_FOUND",
|
||||
"timestamp": "2025-01-12T10:30:00Z"
|
||||
}
|
||||
```
|
||||
|
||||
### Validace dat
|
||||
```json
|
||||
{
|
||||
"currency": "USD",
|
||||
"validation_year": 2025,
|
||||
"adaptive_analysis": {
|
||||
"adaptive_threshold": 1.5,
|
||||
"base_threshold": 1.0,
|
||||
"volatility_percent": 0.24,
|
||||
"data_points": 62
|
||||
},
|
||||
"price_change_violations": [
|
||||
{
|
||||
"date": "06.01.2025",
|
||||
"change_percent": 1.19,
|
||||
"severity": "minor"
|
||||
}
|
||||
],
|
||||
"temporal_gaps": [],
|
||||
"trading_days_validation": {
|
||||
"expected_trading_days": 251,
|
||||
"actual_data_points": 251,
|
||||
"discrepancy_days": 0,
|
||||
"data_completeness_percent": 100.0
|
||||
},
|
||||
"record_counts_by_period": {
|
||||
"2025": {
|
||||
"year": 251,
|
||||
"half_year": {"H1": 124, "H2": 127},
|
||||
"quarter": {"Q1": 63, "Q2": 61, "Q3": 66, "Q4": 61},
|
||||
"month": {"01": 22, "02": 20, "03": 21},
|
||||
"week": {"W01": 5, "W02": 5}
|
||||
}
|
||||
},
|
||||
"data_quality_score": 95
|
||||
}
|
||||
```
|
||||
|
||||
### Počty záznamů podle období
|
||||
```json
|
||||
{
|
||||
"currency": "USD",
|
||||
"record_counts": {
|
||||
"2025": {
|
||||
"year": 251,
|
||||
"half_year": {"H1": 124, "H2": 127},
|
||||
"quarter": {"Q1": 63, "Q2": 61, "Q3": 66, "Q4": 61},
|
||||
"month": {"01": 22, "02": 20, "03": 21, "04": 20},
|
||||
"week": {"W01": 5, "W02": 5, "W03": 5}
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## Chování při různých časech a datumech
|
||||
|
||||
- **Budoucí datum**: Program vrátí chybu, protože kurzy pro budoucí data ještě nebyly vydány.
|
||||
|
||||
508
src/cli.py
508
src/cli.py
@@ -3,25 +3,29 @@
|
||||
import argparse
|
||||
import sys
|
||||
import os
|
||||
import json
|
||||
from datetime import datetime
|
||||
|
||||
# Přidání adresáře src do sys.path, aby bylo možné importovat moduly
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
|
||||
|
||||
import data_fetcher
|
||||
import database
|
||||
import data_fetcher
|
||||
import holidays
|
||||
import rate_finder
|
||||
import rate_reporter
|
||||
import data_validator
|
||||
|
||||
# Global debug flag
|
||||
DEBUG = False
|
||||
|
||||
|
||||
def debug_print(*args, **kwargs):
|
||||
"""Print debug messages only if debug mode is enabled."""
|
||||
if DEBUG:
|
||||
print(*args, **kwargs)
|
||||
|
||||
|
||||
def set_debug_mode(debug):
|
||||
"""Set the debug mode for this module."""
|
||||
global DEBUG
|
||||
@@ -33,6 +37,76 @@ def set_debug_mode(debug):
|
||||
holidays.set_debug_mode(DEBUG)
|
||||
rate_finder.set_debug_mode(DEBUG)
|
||||
rate_reporter.set_debug_mode(DEBUG)
|
||||
data_validator.set_debug_mode(DEBUG)
|
||||
|
||||
|
||||
def format_single_rate_json(
|
||||
currency, rate, requested_date, actual_date=None, fallback=False
|
||||
):
|
||||
"""Format single rate lookup as JSON."""
|
||||
data = {
|
||||
"currency": currency,
|
||||
"rate": float(rate) if rate is not None else None,
|
||||
"date": requested_date,
|
||||
"timestamp": datetime.now().isoformat() + "Z",
|
||||
}
|
||||
if fallback and actual_date:
|
||||
data["actual_date"] = actual_date
|
||||
data["fallback"] = True
|
||||
return data
|
||||
|
||||
|
||||
def format_tax_rate_json(currency, year, rate, monthly_rates=None):
|
||||
"""Format unified tax rate as JSON."""
|
||||
data = {
|
||||
"currency": currency,
|
||||
"year": year,
|
||||
"unified_rate": float(rate) if rate is not None else None,
|
||||
"calculation_date": datetime.now().isoformat() + "Z",
|
||||
}
|
||||
if monthly_rates:
|
||||
data["monthly_rates"] = monthly_rates
|
||||
return data
|
||||
|
||||
|
||||
def format_multi_year_json(currency, year_results):
|
||||
"""Format multi-year stats as JSON."""
|
||||
data = {
|
||||
"currency": currency,
|
||||
"results": [
|
||||
{"year": year, "unified_rate": float(rate) if rate is not None else None}
|
||||
for year, rate in year_results
|
||||
],
|
||||
"calculation_date": datetime.now().isoformat() + "Z",
|
||||
}
|
||||
return data
|
||||
|
||||
|
||||
def format_last_rate_json(currency, rate, date):
|
||||
"""Format last available rate as JSON."""
|
||||
data = {
|
||||
"currency": currency,
|
||||
"rate": float(rate) if rate is not None else None,
|
||||
"date": date,
|
||||
"timestamp": datetime.now().isoformat() + "Z",
|
||||
}
|
||||
return data
|
||||
|
||||
|
||||
def format_error_json(error_msg, error_code=None, details=None):
|
||||
"""Format error response as JSON."""
|
||||
data = {"error": error_msg, "timestamp": datetime.now().isoformat() + "Z"}
|
||||
if error_code:
|
||||
data["code"] = error_code
|
||||
if details:
|
||||
data["details"] = details
|
||||
return data
|
||||
|
||||
|
||||
def output_json(data):
|
||||
"""Output data as formatted JSON."""
|
||||
print(json.dumps(data, indent=2, ensure_ascii=False))
|
||||
|
||||
|
||||
def check_and_update_yearly_data():
|
||||
"""
|
||||
@@ -42,10 +116,14 @@ def check_and_update_yearly_data():
|
||||
debug_print(f"Kontroluji konzistenci ročních dat pro rok {current_year}...")
|
||||
|
||||
# Zkontrolujeme konzistenci dat
|
||||
is_consistent = data_fetcher.check_yearly_data_consistency(current_year, output_dir="data")
|
||||
is_consistent = data_fetcher.check_yearly_data_consistency(
|
||||
current_year, output_dir="data"
|
||||
)
|
||||
|
||||
if not is_consistent:
|
||||
debug_print(f"Roční data pro rok {current_year} nejsou konzistentní. Stahuji aktualizovaná data...")
|
||||
debug_print(
|
||||
f"Roční data pro rok {current_year} nejsou konzistentní. Stahuji aktualizovaná data..."
|
||||
)
|
||||
# Ujistěme se, že adresář data existuje
|
||||
os.makedirs("data", exist_ok=True)
|
||||
# Stáhneme roční data znovu
|
||||
@@ -53,6 +131,7 @@ def check_and_update_yearly_data():
|
||||
else:
|
||||
debug_print(f"Roční data pro rok {current_year} jsou aktuální.")
|
||||
|
||||
|
||||
def main():
|
||||
global DEBUG
|
||||
|
||||
@@ -61,84 +140,101 @@ def main():
|
||||
|
||||
parser = argparse.ArgumentParser(description="Stahování a správa kurzů měn z ČNB.")
|
||||
parser.add_argument(
|
||||
"--year",
|
||||
type=int,
|
||||
help="Rok, pro který se mají stáhnout data (např. 2020)."
|
||||
"--year", type=int, help="Rok, pro který se mají stáhnout data (např. 2020)."
|
||||
)
|
||||
parser.add_argument(
|
||||
"-c", "--currency",
|
||||
"-c",
|
||||
"--currency",
|
||||
type=str,
|
||||
help="Kód měny (např. USD) pro měsíční stahování, vyhledání kurzu nebo generování reportu."
|
||||
help="Kód měny (např. USD) pro měsíční stahování, vyhledání kurzu nebo generování reportu.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--start-date",
|
||||
type=str,
|
||||
help="Počáteční datum pro měsíční stahování nebo generování reportu ve formátu DD.MM.YYYY."
|
||||
help="Počáteční datum pro měsíční stahování nebo generování reportu ve formátu DD.MM.YYYY.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--end-date",
|
||||
type=str,
|
||||
help="Koncové datum pro měsíční stahování nebo generování reportu ve formátu DD.MM.YYYY."
|
||||
help="Koncové datum pro měsíční stahování nebo generování reportu ve formátu DD.MM.YYYY.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--date",
|
||||
type=str,
|
||||
help="Datum pro stažení denních kurzů ve formátu DD.MM.YYYY."
|
||||
help="Datum pro stažení denních kurzů ve formátu DD.MM.YYYY.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--get-rate", "-d",
|
||||
"--get-rate",
|
||||
"-d",
|
||||
type=str,
|
||||
help="Vyhledá kurz pro zadané datum. Formát: DD.MM.YYYY"
|
||||
help="Vyhledá kurz pro zadané datum. Formát: DD.MM.YYYY",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--auto-download",
|
||||
action="store_true",
|
||||
help="Automaticky stáhne denní data, pokud je po 14:30 a kurz pro dnešní datum není k dispozici."
|
||||
help="Automaticky stáhne denní data, pokud je po 14:30 a kurz pro dnešní datum není k dispozici.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--report-year",
|
||||
type=int,
|
||||
help="Rok, pro který se má vygenerovat report kurzů."
|
||||
"--report-year", type=int, help="Rok, pro který se má vygenerovat report kurzů."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--report-month",
|
||||
type=int,
|
||||
help="Měsíc, pro který se má vygenerovat report kurzů (1-12). Vyžaduje --report-year."
|
||||
help="Měsíc, pro který se má vygenerovat report kurzů (1-12). Vyžaduje --report-year.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--report-period",
|
||||
nargs=2,
|
||||
metavar=('START_DATE', 'END_DATE'),
|
||||
help="Období, pro které se má vygenerovat report kurzů. Formát: DD.MM.YYYY DD.MM.YYYY"
|
||||
metavar=("START_DATE", "END_DATE"),
|
||||
help="Období, pro které se má vygenerovat report kurzů. Formát: DD.MM.YYYY DD.MM.YYYY",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--stats",
|
||||
nargs='?',
|
||||
nargs="?",
|
||||
const=True,
|
||||
type=int,
|
||||
help="Vygeneruje 'Jednotný kurz' pro daňové účely podle metodiky ČNB. "
|
||||
"Pokud je zadán rok, vytvoří kurz pro konkrétní rok. "
|
||||
"Pokud není rok zadán, vytvoří kurzy pro všechny roky s dostupnými daty."
|
||||
"Pokud není rok zadán, vytvoří kurzy pro všechny roky s dostupnými daty.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--debug",
|
||||
"--validate",
|
||||
action="store_true",
|
||||
help="Zobrazí podrobné ladicí informace."
|
||||
help="Validuje data pro měnu nebo všechny měny. Zkontroluje konzistenci kurzů a detekuje možné chyby.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--record-counts",
|
||||
action="store_true",
|
||||
help="Zobrazí počet záznamů podle časových období (týden, měsíc, čtvrtletí, pololetí, rok).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--change-threshold",
|
||||
type=float,
|
||||
default=1.0,
|
||||
help="Práh pro detekci změn kurzů v procentech (výchozí: 1.0).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--gap-threshold",
|
||||
type=int,
|
||||
default=3,
|
||||
help="Maximální přijatelná mezera v pracovních dnech (výchozí: 3).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--debug", action="store_true", help="Zobrazí podrobné ladicí informace."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--json",
|
||||
action="store_true",
|
||||
help="Výstup ve formátu JSON místo prostého textu pro programové zpracování.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--no-adaptive",
|
||||
action="store_true",
|
||||
help="Vypne adaptivní učení prahů na základě historických dat.",
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# Pokud nebyly zadány žádné argumenty, vytiskneme nápovědu a seznam dostupných měn
|
||||
if len(sys.argv) == 1:
|
||||
parser.print_help()
|
||||
print("\nDostupné měny:")
|
||||
currencies = database.get_available_currencies()
|
||||
if currencies:
|
||||
print(", ".join(currencies))
|
||||
else:
|
||||
print("Žádné měny nejsou v databázi k dispozici.")
|
||||
sys.exit(0)
|
||||
|
||||
# Nastavíme debug mód
|
||||
DEBUG = args.debug
|
||||
set_debug_mode(DEBUG)
|
||||
@@ -152,131 +248,222 @@ def main():
|
||||
check_and_update_yearly_data()
|
||||
else:
|
||||
# V normálním módu zkontrolujeme pouze při stahování dat
|
||||
if args.year or args.start_date or args.end_date or args.date or args.get_rate or args.report_year or args.report_period or args.stats:
|
||||
if (
|
||||
args.year
|
||||
or args.start_date
|
||||
or args.end_date
|
||||
or args.date
|
||||
or args.get_rate
|
||||
or args.report_year
|
||||
or args.report_period
|
||||
or args.stats
|
||||
):
|
||||
current_year = datetime.now().year
|
||||
# Pro jednoduchost v normálním módu nebudeme kontrolovat konzistenci automaticky
|
||||
pass
|
||||
|
||||
# Zde bude logika pro zpracování argumentů
|
||||
if args.year:
|
||||
debug_print(f"Stahuji roční data pro rok {args.year}...")
|
||||
# Ujistěme se, že adresář data existuje
|
||||
os.makedirs("data", exist_ok=True)
|
||||
# Volání funkce pro stažení ročních dat
|
||||
data_fetcher.download_yearly_data(args.year, output_dir="data")
|
||||
elif args.currency and args.start_date and args.end_date and not args.report_period:
|
||||
# Zde bude logika pro zpracování argumentů
|
||||
if args.validate:
|
||||
# Validation command
|
||||
base_threshold = args.change_threshold
|
||||
adaptive = not args.no_adaptive
|
||||
max_gap_days = getattr(args, "gap_threshold", 3) # Default to 3 if not defined
|
||||
|
||||
if args.currency:
|
||||
# Validate specific currency
|
||||
debug_print(f"Validuji data pro měnu {args.currency}...")
|
||||
results = data_validator.validate_currency_data(
|
||||
args.currency, args.year, base_threshold, adaptive, max_gap_days
|
||||
)
|
||||
|
||||
if args.json:
|
||||
output_json(results)
|
||||
else:
|
||||
text_output = data_validator.format_validation_text(results)
|
||||
print(text_output)
|
||||
else:
|
||||
# Validate all currencies
|
||||
debug_print("Validuji data pro všechny měny...")
|
||||
results = data_validator.validate_all_currencies(
|
||||
args.year, base_threshold, adaptive, max_gap_days
|
||||
)
|
||||
|
||||
if args.json:
|
||||
output_json(results)
|
||||
else:
|
||||
text_output = data_validator.format_validation_text(results)
|
||||
print(text_output)
|
||||
elif args.record_counts:
|
||||
# Record counts command
|
||||
if not args.currency:
|
||||
print(
|
||||
"Chyba: Pro --record-counts je nutné zadat měnu pomocí -c/--currency."
|
||||
)
|
||||
sys.exit(1)
|
||||
|
||||
debug_print(f"Získávám počty záznamů pro měnu {args.currency}...")
|
||||
record_counts = data_validator.get_record_counts_by_period(
|
||||
args.currency, args.year
|
||||
)
|
||||
|
||||
if args.json:
|
||||
output_json({"currency": args.currency, "record_counts": record_counts})
|
||||
else:
|
||||
print(f"Record Counts for {args.currency}:")
|
||||
print("=" * 50)
|
||||
|
||||
for year_key, periods in record_counts.items():
|
||||
print(f"\nYear {year_key}:")
|
||||
print(f" Total records: {periods.get('year', 0)}")
|
||||
|
||||
# Half years
|
||||
half_years = periods.get("half_year", {})
|
||||
if half_years:
|
||||
print(
|
||||
f" Half years: H1={half_years.get('H1', 0)}, H2={half_years.get('H2', 0)}"
|
||||
)
|
||||
|
||||
# Quarters
|
||||
quarters = periods.get("quarter", {})
|
||||
if quarters:
|
||||
quarter_str = ", ".join(
|
||||
[f"Q{q}={quarters.get(f'Q{q}', 0)}" for q in range(1, 5)]
|
||||
)
|
||||
print(f" Quarters: {quarter_str}")
|
||||
|
||||
# Months
|
||||
months = periods.get("month", {})
|
||||
if months:
|
||||
month_list = []
|
||||
for month in range(1, 13):
|
||||
month_key = f"{month:02d}"
|
||||
count = months.get(month_key, 0)
|
||||
month_list.append(f"{month}={count}")
|
||||
print(f" Months: {', '.join(month_list)}")
|
||||
|
||||
# Weeks summary
|
||||
weeks = periods.get("week", {})
|
||||
if weeks:
|
||||
total_weeks = len(weeks)
|
||||
if total_weeks <= 10:
|
||||
week_list = sorted([f"{w}={weeks[w]}" for w in weeks.keys()])
|
||||
print(f" Weeks: {', '.join(week_list)}")
|
||||
else:
|
||||
sample_weeks = sorted(list(weeks.keys())[:5])
|
||||
week_sample = [f"{w}={weeks[w]}" for w in sample_weeks]
|
||||
print(
|
||||
f" Weeks: {', '.join(week_sample)}... ({total_weeks} total weeks)"
|
||||
)
|
||||
elif args.year:
|
||||
# Validation command
|
||||
base_threshold = args.change_threshold
|
||||
adaptive = not args.no_adaptive
|
||||
|
||||
if args.currency:
|
||||
# Validate specific currency
|
||||
debug_print(f"Validuji data pro měnu {args.currency}...")
|
||||
results = data_validator.validate_currency_data(
|
||||
args.currency, args.year, base_threshold, adaptive
|
||||
)
|
||||
|
||||
if args.json:
|
||||
output_json(results)
|
||||
else:
|
||||
text_output = data_validator.format_validation_text(results)
|
||||
print(text_output)
|
||||
else:
|
||||
# Validate all currencies
|
||||
debug_print("Validuji data pro všechny měny...")
|
||||
results = data_validator.validate_all_currencies(
|
||||
args.year, base_threshold, adaptive
|
||||
)
|
||||
|
||||
if args.json:
|
||||
output_json(results)
|
||||
else:
|
||||
text_output = data_validator.format_validation_text(results)
|
||||
print(text_output)
|
||||
return
|
||||
# elif args.currency and args.start_date and args.end_date and not args.report_period:
|
||||
# Měsíční stahování dat
|
||||
debug_print(f"Stahuji měsíční data pro měnu {args.currency} od {args.start_date} do {args.end_date}...")
|
||||
debug_print("HIT: Monthly download condition")
|
||||
debug_print(
|
||||
f"Stahuji měsíční data pro měnu {args.currency} od {args.start_date} do {args.end_date}..."
|
||||
)
|
||||
# Ujistěme se, že adresář data existuje
|
||||
os.makedirs("data", exist_ok=True)
|
||||
# Volání funkce pro stažení měsíčních dat
|
||||
data_fetcher.download_monthly_data(args.currency, args.start_date, args.end_date, output_dir="data")
|
||||
data_fetcher.download_monthly_data(
|
||||
args.currency, args.start_date, args.end_date, output_dir="data"
|
||||
)
|
||||
elif args.report_period and args.currency:
|
||||
start_date, end_date = args.report_period
|
||||
debug_print("HIT: Report period condition")
|
||||
debug_print(
|
||||
f"Generuji report pro měnu {args.currency} od {start_date} do {end_date}..."
|
||||
)
|
||||
rate_reporter.generate_period_report(
|
||||
start_date, end_date, args.currency, output_dir="data"
|
||||
)
|
||||
elif args.date:
|
||||
debug_print("HIT: Daily data condition")
|
||||
debug_print(f"Stahuji denní data pro datum {args.date}...")
|
||||
# Ujistěme se, že adresář data existuje
|
||||
os.makedirs("data", exist_ok=True)
|
||||
# Volání funkce pro stažení denních dat
|
||||
data_fetcher.download_daily_data(args.date, output_dir="data")
|
||||
elif args.get_rate and args.currency:
|
||||
debug_print("HIT: Get rate condition")
|
||||
date_str = args.get_rate
|
||||
currency_code = args.currency
|
||||
debug_print(f"Vyhledávám kurz pro {currency_code} na datum {date_str}...")
|
||||
rate = rate_finder.get_rate_for_date(date_str, currency_code)
|
||||
if rate:
|
||||
if args.json:
|
||||
json_data = format_single_rate_json(currency_code, rate, date_str)
|
||||
output_json(json_data)
|
||||
else:
|
||||
# Pro --get-rate v normálním režimu zobrazíme pouze kurz
|
||||
if not DEBUG:
|
||||
print(rate)
|
||||
else:
|
||||
print(f"Kurz {currency_code} na datum {date_str} (nebo nejbližší pracovní den): {rate}")
|
||||
print(
|
||||
f"Kurz {currency_code} na datum {date_str} (nebo nejbližší pracovní den): {rate}"
|
||||
)
|
||||
else:
|
||||
# Pokud nebyl kurz nalezen a je aktivní přepínač --auto-download, zkusíme stáhnout denní data
|
||||
if args.auto_download:
|
||||
try:
|
||||
requested_date = datetime.strptime(date_str, "%d.%m.%Y")
|
||||
today = datetime.now()
|
||||
|
||||
# Zkontrolujeme, zda je požadované datum dnešní
|
||||
if requested_date.date() == today.date():
|
||||
# Zkontrolujeme, zda je čas po 14:30
|
||||
if today.time() >= datetime.strptime("14:30", "%H:%M").time():
|
||||
debug_print("Automaticky stahuji denní data...")
|
||||
# Ujistěme se, že adresář data existuje
|
||||
os.makedirs("data", exist_ok=True)
|
||||
# Stáhneme denní data pro dnešní datum
|
||||
today_str = today.strftime("%d.%m.%Y")
|
||||
data_fetcher.download_daily_data(today_str, output_dir="data")
|
||||
# Zkusíme znovu vyhledat kurz
|
||||
rate = rate_finder.get_rate_for_date(date_str, currency_code)
|
||||
if rate:
|
||||
if not DEBUG:
|
||||
print(rate)
|
||||
else:
|
||||
print(f"Kurz {currency_code} na datum {date_str} (nebo nejbližší pracovní den): {rate}")
|
||||
# Rate not found
|
||||
if args.json:
|
||||
error_data = format_error_json(
|
||||
f"Kurz {currency_code} na datum {date_str} nebyl nalezen",
|
||||
"RATE_NOT_FOUND",
|
||||
)
|
||||
output_json(error_data)
|
||||
else:
|
||||
if not DEBUG:
|
||||
print("Kurz nenalezen")
|
||||
else:
|
||||
print(f"Kurz {currency_code} na datum {date_str} (ani v předchozích dnech) nebyl nalezen ani po stažení denních dat.")
|
||||
else:
|
||||
if not DEBUG:
|
||||
print("Kurz nenalezen")
|
||||
else:
|
||||
print(f"Chyba: Automatické stahování nelze provést, protože čas ještě není po 14:30. Aktuální čas je {today.strftime('%H:%M')}.")
|
||||
else:
|
||||
if not DEBUG:
|
||||
print("Kurz nenalezen")
|
||||
else:
|
||||
print("Automatické stahování denních dat je možné pouze pro dnešní datum.")
|
||||
except ValueError:
|
||||
if not DEBUG:
|
||||
print("Kurz nenalezen")
|
||||
else:
|
||||
print(f"Neplatný formát data: {date_str}")
|
||||
else:
|
||||
if not DEBUG:
|
||||
print("Kurz nenalezen")
|
||||
else:
|
||||
print(f"Kurz {currency_code} na datum {date_str} (ani v předchozích dnech) nebyl nalezen.")
|
||||
print(
|
||||
f"Kurz {currency_code} na datum {date_str} (ani v předchozích dnech) nebyl nalezen."
|
||||
)
|
||||
elif args.get_rate is not None and not args.currency:
|
||||
debug_print("HIT: Get rate without currency condition")
|
||||
# Pokud je zadán --get-rate bez data a bez měny
|
||||
if DEBUG:
|
||||
print("Chyba: Pro použití --get-rate musí být zadána měna pomocí -c/--currency.")
|
||||
print(
|
||||
"Chyba: Pro použití --get-rate musí být zadána měna pomocí -c/--currency."
|
||||
)
|
||||
sys.exit(1)
|
||||
elif args.currency and not args.get_rate:
|
||||
# Pokud je zadána měna, ale není zadán --get-rate, vytiskneme poslední dostupný kurz
|
||||
currency_code = args.currency
|
||||
debug_print(f"Vyhledávám poslední dostupný kurz pro {currency_code}...")
|
||||
rate, date = database.get_last_rate_for_currency(currency_code)
|
||||
if rate and date:
|
||||
# Pro normální režim zobrazíme kurz ve formátu "11.11 # dated: dd.mm.yyyy"
|
||||
if not DEBUG:
|
||||
print(f"{rate} # dated: {date}")
|
||||
else:
|
||||
print(f"Poslední dostupný kurz {currency_code}: {rate} # dated: {date}")
|
||||
else:
|
||||
if not DEBUG:
|
||||
print("Kurz nenalezen")
|
||||
else:
|
||||
print(f"Poslední dostupný kurz pro {currency_code} nebyl nalezen.")
|
||||
elif args.report_year and args.currency:
|
||||
if args.report_month:
|
||||
# Generování měsíčního reportu
|
||||
debug_print(f"Generuji měsíční report pro {args.currency} za {args.report_month}/{args.report_year}...")
|
||||
rate_reporter.generate_monthly_report(args.report_year, args.report_month, args.currency, output_dir="data")
|
||||
else:
|
||||
# Generování ročního reportu
|
||||
debug_print(f"Generuji roční report pro {args.currency} za rok {args.report_year}...")
|
||||
rate_reporter.generate_yearly_report(args.report_year, args.currency, output_dir="data")
|
||||
elif args.report_period and args.currency:
|
||||
# Generování reportu za období
|
||||
start_date, end_date = args.report_period
|
||||
debug_print(f"Generuji report pro {args.currency} za období {start_date} - {end_date}...")
|
||||
rate_reporter.generate_period_report(start_date, end_date, args.currency, output_dir="data")
|
||||
# DŮLEŽITÉ: Pořadí následujících elif podmínek je důležité!
|
||||
# Nejprve zpracujeme --stats, pak teprve "poslední dostupný kurz"
|
||||
elif args.stats is not None and args.currency:
|
||||
debug_print("HIT: Stats condition")
|
||||
currency_code = args.currency
|
||||
if args.stats is True:
|
||||
# Pokud je --stats zadán bez roku, vytvoříme kurzy pro všechny roky s dostupnými daty
|
||||
debug_print(f"Generuji 'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} pro všechny roky...")
|
||||
debug_print(
|
||||
f"Generuji 'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} pro všechny roky..."
|
||||
)
|
||||
|
||||
# Získáme seznam všech roků s daty
|
||||
years = database.get_years_with_data()
|
||||
@@ -288,58 +475,115 @@ def main():
|
||||
return
|
||||
|
||||
# Pro každý rok vypočítáme 'Jednotný kurz'
|
||||
year_results = []
|
||||
for year in years:
|
||||
# Zkontrolujeme, zda databáze obsahuje data pro daný rok
|
||||
if not rate_finder.check_year_data_in_db(year):
|
||||
debug_print(f"Databáze neobsahuje data pro rok {year}. Stahuji roční data...")
|
||||
debug_print(
|
||||
f"Databáze neobsahuje data pro rok {year}. Stahuji roční data..."
|
||||
)
|
||||
# Ujistěme se, že adresář data existuje
|
||||
os.makedirs("data", exist_ok=True)
|
||||
# Stáhneme roční data s vynuceným stažením
|
||||
data_fetcher.download_yearly_data(year, output_dir="data", force=True)
|
||||
data_fetcher.download_yearly_data(
|
||||
year, output_dir="data", force=True
|
||||
)
|
||||
|
||||
# Vypočítáme 'Jednotný kurz' podle metodiky ČNB
|
||||
tax_rate = rate_reporter.calculate_tax_yearly_average(year, currency_code, output_dir="data")
|
||||
tax_rate = rate_reporter.calculate_tax_yearly_average(
|
||||
year, currency_code, output_dir="data"
|
||||
)
|
||||
year_results.append((year, tax_rate))
|
||||
|
||||
if not args.json:
|
||||
if tax_rate:
|
||||
# Pro --stats v normálním režimu zobrazíme pouze 'Jednotný kurz' zaokrouhlený na 2 desetinná místa
|
||||
if not DEBUG:
|
||||
print(f"{year}: {tax_rate:.2f}")
|
||||
else:
|
||||
print(f"'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year}: {tax_rate:.2f}")
|
||||
print(
|
||||
f"'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year}: {tax_rate:.2f}"
|
||||
)
|
||||
else:
|
||||
if not DEBUG:
|
||||
print(f"{year}: 'Jednotný kurz' nenalezen")
|
||||
else:
|
||||
print(f"'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year} nebyl nalezen.")
|
||||
print(
|
||||
f"'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year} nebyl nalezen."
|
||||
)
|
||||
|
||||
# Output JSON for multi-year results
|
||||
if args.json:
|
||||
json_data = format_multi_year_json(currency_code, year_results)
|
||||
output_json(json_data)
|
||||
else:
|
||||
# Pokud je --stats zadán s konkrétním rokem
|
||||
year = args.stats
|
||||
debug_print(f"Generuji 'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year}...")
|
||||
debug_print(
|
||||
f"Generuji 'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year}..."
|
||||
)
|
||||
|
||||
# Zkontrolujeme, zda databáze obsahuje data pro daný rok
|
||||
if not rate_finder.check_year_data_in_db(year):
|
||||
debug_print(f"Databáze neobsahuje data pro rok {year}. Stahuji roční data...")
|
||||
debug_print(
|
||||
f"Databáze neobsahuje data pro rok {year}. Stahuji roční data..."
|
||||
)
|
||||
# Ujistěme se, že adresář data existuje
|
||||
os.makedirs("data", exist_ok=True)
|
||||
# Stáhneme roční data s vynuceným stažením
|
||||
data_fetcher.download_yearly_data(year, output_dir="data", force=True)
|
||||
|
||||
# Vypočítáme 'Jednotný kurz' podle metodiky ČNB
|
||||
tax_rate = rate_reporter.calculate_tax_yearly_average(year, currency_code, output_dir="data")
|
||||
tax_rate = rate_reporter.calculate_tax_yearly_average(
|
||||
year, currency_code, output_dir="data"
|
||||
)
|
||||
if args.json:
|
||||
json_data = format_tax_rate_json(currency_code, year, tax_rate)
|
||||
output_json(json_data)
|
||||
else:
|
||||
if tax_rate:
|
||||
# Pro --stats v normálním režimu zobrazíme pouze 'Jednotný kurz' zaokrouhlený na 2 desetinná místa
|
||||
if not DEBUG:
|
||||
print(f"{tax_rate:.2f}")
|
||||
else:
|
||||
print(f"'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year}: {tax_rate:.2f}")
|
||||
print(
|
||||
f"'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year}: {tax_rate:.2f}"
|
||||
)
|
||||
else:
|
||||
if not DEBUG:
|
||||
print("'Jednotný kurz' nenalezen")
|
||||
else:
|
||||
print(f"'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year} nebyl nalezen.")
|
||||
print(
|
||||
f"'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year} nebyl nalezen."
|
||||
)
|
||||
elif args.currency and not args.get_rate:
|
||||
# Pokud je zadána měna, ale není zadán --get-rate, vytiskneme poslední dostupný kurz
|
||||
# Toto musí být až po --stats, jinak by se --stats nikdy nevykonalo
|
||||
currency_code = args.currency
|
||||
debug_print(f"Vyhledávám poslední dostupný kurz pro {currency_code}...")
|
||||
rate, date = database.get_last_rate_for_currency(currency_code)
|
||||
if args.json:
|
||||
json_data = format_last_rate_json(currency_code, rate, date)
|
||||
output_json(json_data)
|
||||
else:
|
||||
if rate and date:
|
||||
# Pro normální režim zobrazíme kurz ve formátu "11.11 # dated: dd.mm.yyyy"
|
||||
if not DEBUG:
|
||||
print(f"{rate} # dated: {date}")
|
||||
else:
|
||||
print(
|
||||
f"Poslední dostupný kurz {currency_code}: {rate} # dated: {date}"
|
||||
)
|
||||
else:
|
||||
if not DEBUG:
|
||||
print("Kurz nenalezen")
|
||||
else:
|
||||
print(f"Poslední dostupný kurz pro {currency_code} nebyl nalezen.")
|
||||
else:
|
||||
if DEBUG:
|
||||
parser.print_help()
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
789
src/data_validator.py
Normal file
789
src/data_validator.py
Normal file
@@ -0,0 +1,789 @@
|
||||
import sys
|
||||
import os
|
||||
import json
|
||||
from datetime import datetime, timedelta
|
||||
from collections import defaultdict
|
||||
import statistics
|
||||
|
||||
# Přidání adresáře src do sys.path, aby bylo možné importovat moduly
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__)))
|
||||
|
||||
import database
|
||||
import holidays
|
||||
|
||||
# Global debug flag
|
||||
DEBUG = False
|
||||
|
||||
|
||||
def debug_print(*args, **kwargs):
|
||||
"""Print debug messages only if debug mode is enabled."""
|
||||
if DEBUG:
|
||||
print(*args, **kwargs)
|
||||
|
||||
|
||||
def set_debug_mode(debug):
|
||||
"""Set the debug mode for this module."""
|
||||
global DEBUG
|
||||
DEBUG = debug
|
||||
|
||||
|
||||
def calculate_adaptive_threshold(currency_code, base_threshold=1.0, learning_months=3):
|
||||
"""
|
||||
Calculates adaptive threshold based on 3-month historical volatility.
|
||||
|
||||
:param currency_code: Currency to analyze
|
||||
:param base_threshold: Base threshold percentage
|
||||
:param learning_months: Months of history to analyze
|
||||
:return: Adaptive threshold and volatility statistics
|
||||
"""
|
||||
try:
|
||||
# Calculate date range for learning (3 months back)
|
||||
end_date = datetime.now()
|
||||
start_date = end_date - timedelta(days=learning_months * 30)
|
||||
|
||||
# Get all rates for the period
|
||||
rates_data = []
|
||||
current_date = start_date
|
||||
|
||||
while current_date <= end_date:
|
||||
date_str = current_date.strftime("%d.%m.%Y")
|
||||
rate = database.get_rate(date_str, currency_code)
|
||||
if rate is not None:
|
||||
rates_data.append((current_date, rate))
|
||||
current_date += timedelta(days=1)
|
||||
|
||||
if len(rates_data) < 10:
|
||||
# Insufficient data, return base threshold
|
||||
return {
|
||||
"adaptive_threshold": base_threshold,
|
||||
"base_threshold": base_threshold,
|
||||
"volatility_percent": 0.0,
|
||||
"data_points": len(rates_data),
|
||||
"sufficient_data": False,
|
||||
}
|
||||
|
||||
# Calculate daily percentage changes
|
||||
changes = []
|
||||
for i in range(1, len(rates_data)):
|
||||
prev_rate = rates_data[i - 1][1]
|
||||
curr_rate = rates_data[i][1]
|
||||
if prev_rate > 0:
|
||||
change_pct = abs((curr_rate - prev_rate) / prev_rate) * 100
|
||||
changes.append(change_pct)
|
||||
|
||||
if not changes:
|
||||
return {
|
||||
"adaptive_threshold": base_threshold,
|
||||
"base_threshold": base_threshold,
|
||||
"volatility_percent": 0.0,
|
||||
"data_points": len(rates_data),
|
||||
"sufficient_data": True,
|
||||
}
|
||||
|
||||
# Calculate volatility metrics
|
||||
std_dev = statistics.stdev(changes)
|
||||
percentile_95 = statistics.quantiles(changes, n=20)[18] # 95th percentile
|
||||
|
||||
# Adaptive threshold formula: more conservative of std_dev and percentile_95th/2
|
||||
volatility_factor = max(std_dev, percentile_95 / 2)
|
||||
|
||||
# Apply bounds (0.5% to 5.0%)
|
||||
adaptive_threshold = base_threshold * (
|
||||
1 + min(max(volatility_factor, 0.5), 5.0)
|
||||
)
|
||||
|
||||
return {
|
||||
"adaptive_threshold": adaptive_threshold,
|
||||
"base_threshold": base_threshold,
|
||||
"volatility_percent": std_dev,
|
||||
"percentile_95": percentile_95,
|
||||
"data_points": len(rates_data),
|
||||
"sufficient_data": True,
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
debug_print(f"Error calculating adaptive threshold: {e}")
|
||||
return {
|
||||
"adaptive_threshold": base_threshold,
|
||||
"base_threshold": base_threshold,
|
||||
"volatility_percent": 0.0,
|
||||
"data_points": 0,
|
||||
"sufficient_data": False,
|
||||
"error": str(e),
|
||||
}
|
||||
|
||||
|
||||
def calculate_working_days_gap(start_date, end_date):
|
||||
"""
|
||||
Calculate the number of working days (excluding weekends and holidays) between two dates.
|
||||
|
||||
:param start_date: Start date (datetime)
|
||||
:param end_date: End date (datetime)
|
||||
:return: Number of working days between the dates (exclusive)
|
||||
"""
|
||||
working_days = 0
|
||||
current = start_date + timedelta(days=1) # Start from day after start_date
|
||||
|
||||
while current < end_date:
|
||||
date_str = current.strftime("%d.%m.%Y")
|
||||
if not holidays.is_weekend(date_str) and not holidays.is_holiday(date_str):
|
||||
working_days += 1
|
||||
current += timedelta(days=1)
|
||||
|
||||
return working_days
|
||||
|
||||
|
||||
def calculate_expected_trading_days(year):
|
||||
"""
|
||||
Calculate the expected number of trading days in a year (excluding weekends and holidays).
|
||||
|
||||
:param year: Year to calculate for
|
||||
:return: Dictionary with expected trading days and breakdown
|
||||
"""
|
||||
import calendar
|
||||
|
||||
total_days = 366 if calendar.isleap(year) else 365
|
||||
weekend_days = 0
|
||||
holiday_days = 0
|
||||
|
||||
# Count weekends and holidays
|
||||
for month in range(1, 13):
|
||||
for day in range(1, calendar.monthrange(year, month)[1] + 1):
|
||||
date_str = f"{day:02d}.{month:02d}.{year}"
|
||||
if holidays.is_weekend(date_str):
|
||||
weekend_days += 1
|
||||
elif holidays.is_holiday(date_str):
|
||||
holiday_days += 1
|
||||
|
||||
expected_trading_days = total_days - weekend_days - holiday_days
|
||||
|
||||
return {
|
||||
"total_days": total_days,
|
||||
"weekend_days": weekend_days,
|
||||
"holiday_days": holiday_days,
|
||||
"expected_trading_days": expected_trading_days,
|
||||
}
|
||||
|
||||
|
||||
def validate_trading_days_count(currency_code, year):
|
||||
"""
|
||||
Validate that a year has the appropriate number of trading day entries.
|
||||
|
||||
:param currency_code: Currency to validate
|
||||
:param year: Year to check
|
||||
:return: Validation result with actual vs expected counts
|
||||
"""
|
||||
# Get expected trading days
|
||||
expected = calculate_expected_trading_days(year)
|
||||
|
||||
# Count actual data points for the year
|
||||
actual_count = 0
|
||||
rates_data = []
|
||||
|
||||
start_date = datetime(year, 1, 1)
|
||||
end_date = datetime(year, 12, 31)
|
||||
|
||||
current_date = start_date
|
||||
while current_date <= end_date:
|
||||
date_str = current_date.strftime("%d.%m.%Y")
|
||||
rate = database.get_rate(date_str, currency_code)
|
||||
if rate is not None:
|
||||
actual_count += 1
|
||||
rates_data.append((current_date, rate, date_str))
|
||||
current_date += timedelta(days=1)
|
||||
|
||||
# Calculate discrepancy
|
||||
discrepancy_days = actual_count - expected["expected_trading_days"]
|
||||
discrepancy_percent = (
|
||||
(discrepancy_days / expected["expected_trading_days"]) * 100
|
||||
if expected["expected_trading_days"] > 0
|
||||
else 0
|
||||
)
|
||||
|
||||
# Determine severity
|
||||
severity = "ok"
|
||||
if abs(discrepancy_percent) > 15:
|
||||
severity = "severe"
|
||||
elif abs(discrepancy_percent) > 5:
|
||||
severity = "moderate"
|
||||
elif abs(discrepancy_percent) > 0:
|
||||
severity = "minor"
|
||||
|
||||
return {
|
||||
"expected_trading_days": expected["expected_trading_days"],
|
||||
"actual_data_points": actual_count,
|
||||
"discrepancy_days": discrepancy_days,
|
||||
"discrepancy_percent": round(discrepancy_percent, 2),
|
||||
"severity": severity,
|
||||
"total_days": expected["total_days"],
|
||||
"weekend_days_excluded": expected["weekend_days"],
|
||||
"holiday_days_excluded": expected["holiday_days"],
|
||||
"data_completeness_percent": round(
|
||||
(actual_count / expected["expected_trading_days"]) * 100, 1
|
||||
)
|
||||
if expected["expected_trading_days"] > 0
|
||||
else 0,
|
||||
}
|
||||
|
||||
|
||||
def get_record_counts_by_period(currency_code, year=None):
|
||||
"""
|
||||
Get record counts for different time periods.
|
||||
|
||||
:param currency_code: Currency to analyze
|
||||
:param year: Optional year filter
|
||||
:return: Dictionary with counts by period
|
||||
"""
|
||||
if year:
|
||||
years_to_check = [year]
|
||||
else:
|
||||
years_to_check = database.get_years_with_data()
|
||||
if not years_to_check:
|
||||
return {}
|
||||
|
||||
results = {}
|
||||
|
||||
for check_year in years_to_check:
|
||||
year_results = {}
|
||||
|
||||
# Get all data for the year
|
||||
data_points = []
|
||||
start_date = datetime(check_year, 1, 1)
|
||||
end_date = datetime(check_year, 12, 31)
|
||||
|
||||
current_date = start_date
|
||||
while current_date <= end_date:
|
||||
date_str = current_date.strftime("%d.%m.%Y")
|
||||
rate = database.get_rate(date_str, currency_code)
|
||||
if rate is not None:
|
||||
data_points.append((current_date, rate))
|
||||
current_date += timedelta(days=1)
|
||||
|
||||
# Count by different periods
|
||||
period_counts = {
|
||||
"year": len(data_points),
|
||||
"half_year": {},
|
||||
"quarter": {},
|
||||
"month": {},
|
||||
"week": {},
|
||||
}
|
||||
|
||||
# Half years
|
||||
period_counts["half_year"]["H1"] = len(
|
||||
[d for d in data_points if d[0].month <= 6]
|
||||
)
|
||||
period_counts["half_year"]["H2"] = len(
|
||||
[d for d in data_points if d[0].month > 6]
|
||||
)
|
||||
|
||||
# Quarters
|
||||
for quarter in range(1, 5):
|
||||
start_month = (quarter - 1) * 3 + 1
|
||||
end_month = quarter * 3
|
||||
period_counts["quarter"][f"Q{quarter}"] = len(
|
||||
[d for d in data_points if start_month <= d[0].month <= end_month]
|
||||
)
|
||||
|
||||
# Months
|
||||
for month in range(1, 13):
|
||||
period_counts["month"][f"{month:02d}"] = len(
|
||||
[d for d in data_points if d[0].month == month]
|
||||
)
|
||||
|
||||
# Weeks (approximate by week number)
|
||||
week_counts = {}
|
||||
for data_point in data_points:
|
||||
week_num = data_point[0].isocalendar()[1]
|
||||
week_key = f"W{week_num:02d}"
|
||||
week_counts[week_key] = week_counts.get(week_key, 0) + 1
|
||||
period_counts["week"] = week_counts
|
||||
|
||||
results[str(check_year)] = period_counts
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def detect_temporal_gaps(currency_code, year=None, max_gap_days=3):
|
||||
"""
|
||||
Detect temporal gaps in data sequence (missing working days).
|
||||
|
||||
:param currency_code: Currency to validate
|
||||
:param year: Optional year filter
|
||||
:param max_gap_days: Maximum acceptable working days gap
|
||||
:return: List of gap violations
|
||||
"""
|
||||
gaps = []
|
||||
|
||||
try:
|
||||
# Get all dates and rates for the currency/year
|
||||
rates_data = []
|
||||
if year:
|
||||
# Specific year
|
||||
start_date = datetime(year, 1, 1)
|
||||
end_date = datetime(year, 12, 31)
|
||||
else:
|
||||
# All available data
|
||||
years_with_data = database.get_years_with_data()
|
||||
if not years_with_data:
|
||||
return gaps
|
||||
start_year = min(years_with_data)
|
||||
end_year = max(years_with_data)
|
||||
start_date = datetime(start_year, 1, 1)
|
||||
end_date = datetime(end_year, 12, 31)
|
||||
|
||||
current_date = start_date
|
||||
while current_date <= datetime.now() and current_date <= end_date:
|
||||
date_str = current_date.strftime("%d.%m.%Y")
|
||||
rate = database.get_rate(date_str, currency_code)
|
||||
if rate is not None:
|
||||
rates_data.append((current_date, rate, date_str))
|
||||
current_date += timedelta(days=1)
|
||||
|
||||
# Check for gaps between consecutive data points
|
||||
for i in range(1, len(rates_data)):
|
||||
prev_date, _, prev_date_str = rates_data[i - 1]
|
||||
curr_date, _, curr_date_str = rates_data[i]
|
||||
|
||||
# Calculate working days gap
|
||||
working_days_gap = calculate_working_days_gap(prev_date, curr_date)
|
||||
|
||||
if working_days_gap > max_gap_days:
|
||||
# Determine severity
|
||||
severity = "minor"
|
||||
if working_days_gap > max_gap_days * 3:
|
||||
severity = "severe"
|
||||
elif working_days_gap > max_gap_days * 2:
|
||||
severity = "moderate"
|
||||
|
||||
gap = {
|
||||
"start_date": prev_date_str,
|
||||
"end_date": curr_date_str,
|
||||
"working_days_missing": working_days_gap,
|
||||
"severity": severity,
|
||||
"max_expected_gap": max_gap_days,
|
||||
"recommendation": f"Check data source for {working_days_gap} missing working days",
|
||||
}
|
||||
gaps.append(gap)
|
||||
|
||||
except Exception as e:
|
||||
debug_print(f"Error detecting temporal gaps: {e}")
|
||||
|
||||
return gaps
|
||||
|
||||
|
||||
def detect_price_change_violations(
|
||||
currency_code, year=None, base_threshold=1.0, adaptive=True
|
||||
):
|
||||
"""
|
||||
Detects price changes exceeding thresholds.
|
||||
|
||||
:param currency_code: Currency to validate
|
||||
:param year: Optional year filter
|
||||
:param base_threshold: Base threshold percentage
|
||||
:param adaptive: Whether to use adaptive threshold
|
||||
:return: List of violations
|
||||
"""
|
||||
violations = []
|
||||
|
||||
# Initialize adaptive_info in case of early exception
|
||||
adaptive_info = {
|
||||
"adaptive_threshold": base_threshold,
|
||||
"base_threshold": base_threshold,
|
||||
"volatility_percent": 0.0,
|
||||
"sufficient_data": True,
|
||||
}
|
||||
|
||||
try:
|
||||
# Get adaptive threshold if enabled
|
||||
if adaptive:
|
||||
adaptive_info = calculate_adaptive_threshold(currency_code, base_threshold)
|
||||
|
||||
effective_threshold = adaptive_info["adaptive_threshold"]
|
||||
|
||||
# Get all dates and rates for the currency/year
|
||||
rates_data = []
|
||||
if year:
|
||||
# Specific year
|
||||
start_date = datetime(year, 1, 1)
|
||||
end_date = datetime(year, 12, 31)
|
||||
else:
|
||||
# All available data
|
||||
years_with_data = database.get_years_with_data()
|
||||
if not years_with_data:
|
||||
return violations, adaptive_info
|
||||
start_year = min(years_with_data)
|
||||
end_year = max(years_with_data)
|
||||
start_date = datetime(start_year, 1, 1)
|
||||
end_date = datetime(end_year, 12, 31)
|
||||
|
||||
current_date = start_date
|
||||
while current_date <= datetime.now() and current_date <= end_date:
|
||||
date_str = current_date.strftime("%d.%m.%Y")
|
||||
rate = database.get_rate(date_str, currency_code)
|
||||
if rate is not None:
|
||||
rates_data.append((current_date, rate, date_str))
|
||||
current_date += timedelta(days=1)
|
||||
|
||||
# Check consecutive pairs
|
||||
for i in range(1, len(rates_data)):
|
||||
prev_date, prev_rate, prev_date_str = rates_data[i - 1]
|
||||
curr_date, curr_rate, curr_date_str = rates_data[i]
|
||||
|
||||
if prev_rate > 0:
|
||||
change_pct = abs((curr_rate - prev_rate) / prev_rate) * 100
|
||||
|
||||
# Determine severity
|
||||
severity = "minor"
|
||||
if change_pct > effective_threshold * 3:
|
||||
severity = "severe"
|
||||
elif change_pct > effective_threshold:
|
||||
severity = "moderate"
|
||||
|
||||
# Flag if exceeds base threshold (always) or adaptive threshold
|
||||
if change_pct > base_threshold:
|
||||
violation = {
|
||||
"date": curr_date_str,
|
||||
"previous_date": prev_date_str,
|
||||
"previous_rate": float(prev_rate),
|
||||
"current_rate": float(curr_rate),
|
||||
"change_percent": round(change_pct, 2),
|
||||
"severity": severity,
|
||||
"threshold_exceeded": "adaptive"
|
||||
if change_pct > effective_threshold
|
||||
else "base",
|
||||
"effective_threshold": effective_threshold,
|
||||
}
|
||||
|
||||
# Add corruption risk assessment for severe cases
|
||||
if severity == "severe":
|
||||
violation["corruption_risk"] = "high"
|
||||
violation["recommendation"] = (
|
||||
"Verify data source - potential currency mismatch or data corruption"
|
||||
)
|
||||
|
||||
violations.append(violation)
|
||||
|
||||
except Exception as e:
|
||||
debug_print(f"Error detecting price changes: {e}")
|
||||
|
||||
return violations, adaptive_info
|
||||
|
||||
|
||||
def validate_currency_data(
|
||||
currency_code, year=None, base_threshold=1.0, adaptive=True, max_gap_days=3
|
||||
):
|
||||
"""
|
||||
Comprehensive validation for a currency.
|
||||
|
||||
:param currency_code: Currency to validate
|
||||
:param year: Optional year filter
|
||||
:param base_threshold: Base threshold for price changes
|
||||
:param adaptive: Whether to use adaptive thresholds
|
||||
:param max_gap_days: Maximum acceptable working days gap
|
||||
:return: Validation results
|
||||
"""
|
||||
results = {
|
||||
"currency": currency_code,
|
||||
"validation_year": year,
|
||||
"validation_date": datetime.now().isoformat() + "Z",
|
||||
}
|
||||
|
||||
try:
|
||||
# Price change violations
|
||||
violations, adaptive_info = detect_price_change_violations(
|
||||
currency_code, year, base_threshold, adaptive
|
||||
)
|
||||
|
||||
# Temporal gaps
|
||||
gaps = detect_temporal_gaps(currency_code, year, max_gap_days)
|
||||
|
||||
# Trading days validation
|
||||
trading_days_validation = None
|
||||
if year:
|
||||
trading_days_validation = validate_trading_days_count(currency_code, year)
|
||||
|
||||
# Record counts by period
|
||||
record_counts = get_record_counts_by_period(currency_code, year)
|
||||
|
||||
results["adaptive_analysis"] = adaptive_info
|
||||
results["price_change_violations"] = violations
|
||||
results["temporal_gaps"] = gaps
|
||||
results["trading_days_validation"] = trading_days_validation
|
||||
results["record_counts_by_period"] = record_counts
|
||||
|
||||
# Summary statistics
|
||||
severity_counts = defaultdict(int)
|
||||
for v in violations:
|
||||
severity_counts[v["severity"]] += 1
|
||||
|
||||
gap_severity_counts = defaultdict(int)
|
||||
for g in gaps:
|
||||
gap_severity_counts[g["severity"]] += 1
|
||||
|
||||
results["summary"] = {
|
||||
"total_violations": len(violations),
|
||||
"total_gaps": len(gaps),
|
||||
"severity_breakdown": dict(severity_counts),
|
||||
"gap_severity_breakdown": dict(gap_severity_counts),
|
||||
"base_threshold": base_threshold,
|
||||
"adaptive_enabled": adaptive,
|
||||
"max_gap_days": max_gap_days,
|
||||
}
|
||||
|
||||
# Data quality score (enhanced heuristic)
|
||||
quality_penalty = 0
|
||||
if violations:
|
||||
quality_penalty += (
|
||||
len(violations) * 5 + severity_counts.get("severe", 0) * 20
|
||||
)
|
||||
if gaps:
|
||||
quality_penalty += (
|
||||
len(gaps) * 10 + gap_severity_counts.get("severe", 0) * 30
|
||||
)
|
||||
if trading_days_validation and trading_days_validation["severity"] != "ok":
|
||||
severity_penalty = {"minor": 5, "moderate": 15, "severe": 30}
|
||||
quality_penalty += severity_penalty.get(
|
||||
trading_days_validation["severity"], 0
|
||||
)
|
||||
|
||||
results["data_quality_score"] = max(0, 100 - quality_penalty)
|
||||
|
||||
except Exception as e:
|
||||
results["error"] = str(e)
|
||||
results["data_quality_score"] = 0
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def validate_all_currencies(
|
||||
year=None, base_threshold=1.0, adaptive=True, max_gap_days=3
|
||||
):
|
||||
"""
|
||||
Validates all available currencies.
|
||||
|
||||
:param year: Optional year filter
|
||||
:param base_threshold: Base threshold for price changes
|
||||
:param adaptive: Whether to use adaptive thresholds
|
||||
:param max_gap_days: Maximum acceptable working days gap
|
||||
:return: Validation results for all currencies
|
||||
"""
|
||||
results = {
|
||||
"validation_type": "all_currencies",
|
||||
"validation_year": year,
|
||||
"base_threshold": base_threshold,
|
||||
"adaptive_enabled": adaptive,
|
||||
"max_gap_days": max_gap_days,
|
||||
"validation_date": datetime.now().isoformat() + "Z",
|
||||
"currency_results": [],
|
||||
}
|
||||
|
||||
try:
|
||||
# Get all available currencies (we'll check a few known ones and any in database)
|
||||
currencies_to_check = ["USD", "EUR", "GBP", "CHF", "JPY"]
|
||||
|
||||
for currency in currencies_to_check:
|
||||
try:
|
||||
currency_result = validate_currency_data(
|
||||
currency, year, base_threshold, adaptive, max_gap_days
|
||||
)
|
||||
results["currency_results"].append(currency_result)
|
||||
except Exception as e:
|
||||
results["currency_results"].append(
|
||||
{"currency": currency, "error": str(e)}
|
||||
)
|
||||
|
||||
# Overall summary
|
||||
total_violations = sum(
|
||||
r.get("summary", {}).get("total_violations", 0)
|
||||
for r in results["currency_results"]
|
||||
if "summary" in r
|
||||
)
|
||||
total_gaps = sum(
|
||||
r.get("summary", {}).get("total_gaps", 0)
|
||||
for r in results["currency_results"]
|
||||
if "summary" in r
|
||||
)
|
||||
severe_violations = sum(
|
||||
r.get("summary", {}).get("severity_breakdown", {}).get("severe", 0)
|
||||
for r in results["currency_results"]
|
||||
if "summary" in r
|
||||
)
|
||||
severe_gaps = sum(
|
||||
r.get("summary", {}).get("gap_severity_breakdown", {}).get("severe", 0)
|
||||
for r in results["currency_results"]
|
||||
if "summary" in r
|
||||
)
|
||||
|
||||
results["overall_summary"] = {
|
||||
"currencies_checked": len(results["currency_results"]),
|
||||
"total_violations": total_violations,
|
||||
"total_gaps": total_gaps,
|
||||
"severe_violations": severe_violations,
|
||||
"severe_gaps": severe_gaps,
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
results["error"] = str(e)
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def format_validation_text(results):
|
||||
"""Format validation results as text output."""
|
||||
output = []
|
||||
|
||||
if "currency" in results:
|
||||
# Single currency validation
|
||||
output.append(
|
||||
f"Currency Validation: {results['currency']} ({results.get('validation_year', 'All Years')})"
|
||||
)
|
||||
output.append("=" * 60)
|
||||
|
||||
adaptive = results.get("adaptive_analysis", {})
|
||||
if adaptive.get("sufficient_data", False):
|
||||
output.append("\nAdaptive Analysis (3-month history):")
|
||||
output.append(
|
||||
f"- Historical volatility: {adaptive.get('volatility_percent', 0):.1f}% std dev"
|
||||
)
|
||||
output.append(
|
||||
f"- Adaptive threshold: {adaptive.get('adaptive_threshold', 1.0):.1f}% (base: {adaptive.get('base_threshold', 1.0)}%)"
|
||||
)
|
||||
output.append(f"- Data points analyzed: {adaptive.get('data_points', 0)}")
|
||||
else:
|
||||
output.append(
|
||||
f"\nAdaptive Analysis: Insufficient data (using base threshold: {adaptive.get('base_threshold', 1.0)}%)"
|
||||
)
|
||||
|
||||
violations = results.get("price_change_violations", [])
|
||||
if violations:
|
||||
output.append("\nPrice Change Violations:")
|
||||
for i, v in enumerate(violations, 1):
|
||||
severity = v["severity"].upper()
|
||||
output.append(
|
||||
f"{i}. [{severity}] {v['date']}: {v['previous_rate']:.2f} → {v['current_rate']:.2f} ({'+' if v['change_percent'] > 0 else ''}{v['change_percent']:.2f}%)"
|
||||
)
|
||||
if "recommendation" in v:
|
||||
output.append(f" → {v['recommendation']}")
|
||||
else:
|
||||
output.append("\nPrice Change Violations: None found")
|
||||
|
||||
gaps = results.get("temporal_gaps", [])
|
||||
if gaps:
|
||||
output.append("\nTemporal Gaps:")
|
||||
for i, g in enumerate(gaps, 1):
|
||||
severity = g["severity"].upper()
|
||||
output.append(
|
||||
f"{i}. [{severity}] {g['start_date']} → {g['end_date']}: {g['working_days_missing']} working days missing"
|
||||
)
|
||||
if "recommendation" in g:
|
||||
output.append(f" → {g['recommendation']}")
|
||||
else:
|
||||
output.append("\nTemporal Gaps: None found")
|
||||
|
||||
# Trading days validation
|
||||
trading_validation = results.get("trading_days_validation")
|
||||
if trading_validation:
|
||||
output.append("\nTrading Days Validation:")
|
||||
output.append(
|
||||
f"- Expected trading days: {trading_validation['expected_trading_days']} ({trading_validation.get('total_days', 'N/A')} total - {trading_validation.get('weekend_days_excluded', 0)} weekends - {trading_validation.get('holiday_days_excluded', 0)} holidays)"
|
||||
)
|
||||
output.append(
|
||||
f"- Actual data points: {trading_validation['actual_data_points']}"
|
||||
)
|
||||
output.append(
|
||||
f"- Discrepancy: {trading_validation['discrepancy_days']} days ({trading_validation['discrepancy_percent']}%)"
|
||||
)
|
||||
output.append(
|
||||
f"- Data completeness: {trading_validation['data_completeness_percent']}%"
|
||||
)
|
||||
output.append(f"- Status: {trading_validation['severity'].upper()}")
|
||||
|
||||
# Record counts by period
|
||||
record_counts = results.get("record_counts_by_period", {})
|
||||
if record_counts:
|
||||
for year_key, periods in record_counts.items():
|
||||
output.append(f"\nRecord Counts for {year_key}:")
|
||||
output.append(f"- Year total: {periods.get('year', 0)} records")
|
||||
|
||||
# Half years
|
||||
half_years = periods.get("half_year", {})
|
||||
if half_years:
|
||||
output.append(
|
||||
f"- Half years: H1={half_years.get('H1', 0)}, H2={half_years.get('H2', 0)}"
|
||||
)
|
||||
|
||||
# Quarters
|
||||
quarters = periods.get("quarter", {})
|
||||
if quarters:
|
||||
quarter_str = ", ".join(
|
||||
[f"Q{q}={quarters.get(f'Q{q}', 0)}" for q in range(1, 5)]
|
||||
)
|
||||
output.append(f"- Quarters: {quarter_str}")
|
||||
|
||||
# Months summary
|
||||
months = periods.get("month", {})
|
||||
if months:
|
||||
month_list = [
|
||||
f"{m}={months.get(f'{int(m):02d}', 0)}"
|
||||
for m in [
|
||||
"01",
|
||||
"02",
|
||||
"03",
|
||||
"04",
|
||||
"05",
|
||||
"06",
|
||||
"07",
|
||||
"08",
|
||||
"09",
|
||||
"10",
|
||||
"11",
|
||||
"12",
|
||||
]
|
||||
]
|
||||
output.append(f"- Months: {', '.join(month_list)}")
|
||||
|
||||
# Weeks summary (show first few and indicate total)
|
||||
weeks = periods.get("week", {})
|
||||
if weeks:
|
||||
total_weeks = len(weeks)
|
||||
if total_weeks <= 10:
|
||||
week_list = [f"{w}={weeks[w]}" for w in sorted(weeks.keys())]
|
||||
output.append(f"- Weeks: {', '.join(week_list)}")
|
||||
else:
|
||||
sample_weeks = sorted(list(weeks.keys())[:5])
|
||||
week_sample = [f"{w}={weeks[w]}" for w in sample_weeks]
|
||||
output.append(
|
||||
f"- Weeks: {', '.join(week_sample)}... ({total_weeks} total weeks)"
|
||||
)
|
||||
|
||||
summary = results.get("summary", {})
|
||||
quality_score = results.get("data_quality_score", 0)
|
||||
output.append(f"\nData Quality Score: {quality_score}%")
|
||||
output.append(f"Total violations: {summary.get('total_violations', 0)}")
|
||||
output.append(f"Total gaps: {summary.get('total_gaps', 0)}")
|
||||
|
||||
elif "currency_results" in results:
|
||||
# Multi-currency validation
|
||||
output.append("Multi-Currency Validation Report")
|
||||
output.append("=" * 60)
|
||||
|
||||
for currency_result in results["currency_results"]:
|
||||
currency = currency_result.get("currency", "Unknown")
|
||||
violations = currency_result.get("price_change_violations", [])
|
||||
quality_score = currency_result.get("data_quality_score", 0)
|
||||
|
||||
output.append(f"\n{currency}:")
|
||||
output.append(f" - Violations: {len(violations)}")
|
||||
output.append(f" - Quality Score: {quality_score}%")
|
||||
|
||||
if violations:
|
||||
severe_count = sum(1 for v in violations if v["severity"] == "severe")
|
||||
output.append(f" - Severe violations: {severe_count}")
|
||||
|
||||
overall = results.get("overall_summary", {})
|
||||
output.append("\nOverall Summary:")
|
||||
output.append(f"- Currencies checked: {overall.get('currencies_checked', 0)}")
|
||||
output.append(f"- Total violations: {overall.get('total_violations', 0)}")
|
||||
output.append(f"- Severe violations: {overall.get('severe_violations', 0)}")
|
||||
|
||||
return "\n".join(output)
|
||||
@@ -1,5 +1,6 @@
|
||||
import sys
|
||||
import os
|
||||
import time
|
||||
from datetime import datetime, timedelta
|
||||
import calendar
|
||||
|
||||
@@ -14,16 +15,53 @@ import rate_finder
|
||||
# Global debug flag
|
||||
DEBUG = False
|
||||
|
||||
|
||||
def debug_print(*args, **kwargs):
|
||||
"""Print debug messages only if debug mode is enabled."""
|
||||
if DEBUG:
|
||||
print(*args, **kwargs)
|
||||
|
||||
|
||||
def set_debug_mode(debug):
|
||||
"""Set the debug mode for this module."""
|
||||
global DEBUG
|
||||
DEBUG = debug
|
||||
|
||||
|
||||
def get_czech_day_name(date_str):
|
||||
"""
|
||||
Vrátí český název dne v týdnu pro zadané datum.
|
||||
|
||||
:param date_str: Datum ve formátu DD.MM.YYYY
|
||||
:return: Český název dne v týdnu
|
||||
"""
|
||||
try:
|
||||
date_obj = datetime.strptime(date_str, "%d.%m.%Y")
|
||||
# Czech day names
|
||||
czech_days = [
|
||||
"pondělí",
|
||||
"úterý",
|
||||
"středa",
|
||||
"čtvrtek",
|
||||
"pátek",
|
||||
"sobota",
|
||||
"neděle",
|
||||
]
|
||||
return czech_days[date_obj.weekday()]
|
||||
except ValueError:
|
||||
return "neznámý den"
|
||||
|
||||
|
||||
def is_holiday(date_str):
|
||||
"""
|
||||
Zkontroluje, zda je zadané datum státní svátek.
|
||||
|
||||
:param date_str: Datum ve formátu DD.MM.YYYY
|
||||
:return: True pokud je svátek, jinak False
|
||||
"""
|
||||
return holidays.is_holiday(date_str)
|
||||
|
||||
|
||||
def get_rate_for_date_with_fallback(date_str, currency_code):
|
||||
"""
|
||||
Vyhledá kurz pro zadané datum a měnu. Pokud kurz pro dané datum neexistuje,
|
||||
@@ -50,7 +88,7 @@ def get_rate_for_date_with_fallback(date_str, currency_code):
|
||||
# a pro případnou následující sobotu, neděli či státní svátek"
|
||||
#
|
||||
# To znamená:
|
||||
# - Pro víkendy a svátky hledáme kurz zpět v čase podle pravidel ČNB
|
||||
# - Pro víkendy a svátky hledáme kurz zpět v čase
|
||||
# - Pro běžné dny, které nemají kurz, hledáme kurz z posledního pracovního dne před nimi
|
||||
|
||||
# Zkontrolujeme, zda je datum víkend nebo svátek
|
||||
@@ -80,13 +118,212 @@ def get_rate_for_date_with_fallback(date_str, currency_code):
|
||||
check_date_str = current_date.strftime("%d.%m.%Y")
|
||||
|
||||
# Zkontrolujeme, zda je to pracovní den
|
||||
if not holidays.is_weekend(check_date_str) and not holidays.is_holiday(check_date_str):
|
||||
if not holidays.is_weekend(check_date_str) and not holidays.is_holiday(
|
||||
check_date_str
|
||||
):
|
||||
rate = database.get_rate(check_date_str, currency_code)
|
||||
if rate is not None:
|
||||
return rate
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def get_missing_months_for_tax_calculation(year, currency_code):
|
||||
"""
|
||||
Vrátí seznam měsíců, pro které chybí kurzy k posledním dnům pro výpočet 'Jednotného kurzu'.
|
||||
Zahrnuje pouze měsíce, jejichž poslední den je v minulosti (lze stáhnout).
|
||||
|
||||
:param year: Rok k ověření
|
||||
:param currency_code: Kód měny
|
||||
:return: Seznam měsíců (1-12) s chybějícími kurzy
|
||||
"""
|
||||
import calendar
|
||||
|
||||
missing_months = []
|
||||
|
||||
for month in range(1, 13):
|
||||
# Získáme poslední den měsíce
|
||||
last_day = calendar.monthrange(year, month)[1]
|
||||
date_str = f"{last_day:02d}.{month:02d}.{year}"
|
||||
|
||||
# Pokud je datum v budoucnosti, přeskočíme (nelze stáhnout)
|
||||
date_obj = datetime.strptime(date_str, "%d.%m.%Y")
|
||||
current_date = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
|
||||
if date_obj > current_date:
|
||||
continue
|
||||
|
||||
# Zkusíme najít kurz pro dané datum
|
||||
rate = database.get_rate(date_str, currency_code)
|
||||
if rate is None:
|
||||
# Zkusíme dopočítat
|
||||
calculated_rate = get_rate_for_date_with_fallback(date_str, currency_code)
|
||||
if calculated_rate is None:
|
||||
missing_months.append(month)
|
||||
|
||||
return missing_months
|
||||
|
||||
|
||||
def _is_year_complete_for_tax_calculation(year):
|
||||
"""
|
||||
Zkontroluje, zda je rok kompletní pro výpočet 'Jednotného kurzu' podle metodiky ČNB.
|
||||
Rok je kompletní, pokud máme kurzy pro poslední den každého měsíce.
|
||||
|
||||
:param year: Rok k ověření
|
||||
:return: True pokud je rok kompletní, jinak False
|
||||
"""
|
||||
import calendar
|
||||
|
||||
current_year = datetime.now().year
|
||||
current_month = datetime.now().month
|
||||
current_day = datetime.now().day
|
||||
|
||||
# Pokud je rok v budoucnosti, není kompletní
|
||||
if year > current_year:
|
||||
return False
|
||||
|
||||
# Pro všechny roky (včetně minulých) zkontrolujeme, zda máme data pro všechny měsíce
|
||||
for month in range(1, 13):
|
||||
# Získáme poslední den měsíce
|
||||
last_day = calendar.monthrange(year, month)[1]
|
||||
date_str = f"{last_day:02d}.{month:02d}.{year}"
|
||||
|
||||
# Pokud je datum v budoucnosti, nemůže být kompletní
|
||||
date_obj = datetime.strptime(date_str, "%d.%m.%Y")
|
||||
current_date = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
|
||||
if date_obj > current_date:
|
||||
return False
|
||||
|
||||
# Zkusíme najít kurz pro dané datum
|
||||
rate = database.get_rate(date_str, "USD") # Testujeme na USD jako ukázku
|
||||
if rate is None:
|
||||
# Zkusíme dopočítat
|
||||
calculated_rate = get_rate_for_date_with_fallback(date_str, "USD")
|
||||
if calculated_rate is None:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
# Pro aktuální rok zkontrolujeme, zda máme data pro všechny měsíce
|
||||
for month in range(1, 13):
|
||||
# Získáme poslední den měsíce
|
||||
last_day = calendar.monthrange(year, month)[1]
|
||||
date_str = f"{last_day:02d}.{month:02d}.{year}"
|
||||
|
||||
# Pokud je měsíc v budoucnosti, nemůže být kompletní
|
||||
if year == current_year and month > current_month:
|
||||
return False
|
||||
|
||||
# Zkusíme najít kurz pro dané datum
|
||||
rate = database.get_rate(date_str, "USD") # Testujeme na USD jako ukázku
|
||||
if rate is None:
|
||||
# Zkusíme dopočítat
|
||||
calculated_rate = get_rate_for_date_with_fallback(date_str, "USD")
|
||||
if calculated_rate is None:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def _auto_download_missing_monthly_data(year, currency_code, output_dir="data"):
|
||||
"""
|
||||
Automatically download missing monthly data for tax calculation (silent operation).
|
||||
|
||||
:param year: Year to check
|
||||
:param currency_code: Currency code
|
||||
:param output_dir: Output directory
|
||||
"""
|
||||
missing_months = get_missing_months_for_tax_calculation(year, currency_code)
|
||||
if missing_months:
|
||||
debug_print(
|
||||
f"Auto-downloading missing monthly data for {currency_code} {year}: months {', '.join(f'{m:02d}' for m in missing_months)}"
|
||||
)
|
||||
for month in missing_months:
|
||||
start_date = f"01.{month:02d}.{year}"
|
||||
last_day = calendar.monthrange(year, month)[1]
|
||||
end_date = f"{last_day:02d}.{month:02d}.{year}"
|
||||
try:
|
||||
data_fetcher.download_monthly_data(
|
||||
currency_code, start_date, end_date, output_dir=output_dir
|
||||
)
|
||||
# Small delay to be respectful to the API
|
||||
time.sleep(0.5)
|
||||
except Exception as e:
|
||||
debug_print(
|
||||
f"Failed to download data for {currency_code} {month:02d}/{year}: {e}"
|
||||
)
|
||||
|
||||
|
||||
def calculate_tax_yearly_average(year, currency_code, output_dir="data"):
|
||||
"""
|
||||
Vypočítá 'Jednotný kurz' pro daňové účely podle metodiky ČNB.
|
||||
Jedná se o aritmetický průměr kurzů k posledním dnům každého kalendářního měsíce v roce.
|
||||
|
||||
:param year: Rok
|
||||
:param currency_code: Kód měny (např. USD)
|
||||
:param output_dir: Adresář s daty
|
||||
:return: 'Jednotný kurz' jako desetinné číslo, nebo None pokud není k dispozici
|
||||
"""
|
||||
debug_print(
|
||||
f"Vypočítávám 'Jednotný kurz' pro daňové účely podle metodiky ČNB pro {currency_code} za rok {year}..."
|
||||
)
|
||||
|
||||
# Auto-download missing monthly data if needed (silent operation)
|
||||
_auto_download_missing_monthly_data(year, currency_code, output_dir)
|
||||
|
||||
# Zkontrolujeme, zda databáze obsahuje data pro daný rok
|
||||
if not rate_finder.check_year_data_in_db(year):
|
||||
debug_print(f"Databáze neobsahuje data pro rok {year}. Stahuji roční data...")
|
||||
# Stáhneme roční data s vynuceným stažením
|
||||
os.makedirs("data", exist_ok=True)
|
||||
data_fetcher.download_yearly_data(year, output_dir="data", force=True)
|
||||
|
||||
# Získáme seznam posledních dní všech měsíců v roce
|
||||
monthly_rates = []
|
||||
monthly_dates = []
|
||||
|
||||
debug_print(f"Získávám kurzy k posledním dnům všech měsíců v roce {year}...")
|
||||
|
||||
for month in range(1, 13):
|
||||
# Získáme poslední den měsíce
|
||||
last_day = calendar.monthrange(year, month)[1]
|
||||
date_str = f"{last_day:02d}.{month:02d}.{year}"
|
||||
|
||||
# Získáme kurz pro dané datum
|
||||
rate = database.get_rate(date_str, currency_code)
|
||||
if rate is not None:
|
||||
monthly_rates.append(rate)
|
||||
monthly_dates.append(date_str)
|
||||
debug_print(f"Měsíc {month:02d}: {date_str} = {rate}")
|
||||
else:
|
||||
# Kurz nebyl nalezen, zkusíme dopočítat
|
||||
calculated_rate = get_rate_for_date_with_fallback(date_str, currency_code)
|
||||
if calculated_rate is not None:
|
||||
monthly_rates.append(calculated_rate)
|
||||
monthly_dates.append(date_str)
|
||||
debug_print(
|
||||
f"Měsíc {month:02d}: {date_str} = {calculated_rate} (dopočítaný kurz)"
|
||||
)
|
||||
else:
|
||||
debug_print(f"Měsíc {month:02d}: {date_str} = kurz nenalezen")
|
||||
|
||||
debug_print(f"Počet měsíců s kurzy: {len(monthly_rates)}/12")
|
||||
|
||||
# Musíme mít kurzy pro všech 12 měsíců
|
||||
if len(monthly_rates) != 12:
|
||||
debug_print(
|
||||
f"Varování: Nenalezeny kurzy pro všech 12 měsíců ({len(monthly_rates)}/12)"
|
||||
)
|
||||
return None
|
||||
|
||||
# Výpočet aritmetického průměru
|
||||
average = sum(monthly_rates) / len(monthly_rates)
|
||||
debug_print(f"Součet kurzů: {sum(monthly_rates):.6f}")
|
||||
debug_print(f"Počet měsíců: {len(monthly_rates)}")
|
||||
debug_print(f"'Jednotný kurz' pro daňové účely: {average:.6f}")
|
||||
|
||||
return average
|
||||
|
||||
|
||||
def generate_yearly_report(year, currency_code, output_dir="data"):
|
||||
"""
|
||||
Vygeneruje report kurzů pro zadaný rok a měnu.
|
||||
@@ -125,7 +362,7 @@ def generate_yearly_report(year, currency_code, output_dir="data"):
|
||||
filepath = os.path.join(output_dir, filename)
|
||||
|
||||
try:
|
||||
with open(filepath, 'w', newline='', encoding='utf-8') as csvfile:
|
||||
with open(filepath, "w", newline="", encoding="utf-8") as csvfile:
|
||||
# Nové pořadí sloupců: Datum, Kurz, Den v týdnu, Svátek, Poznámka
|
||||
csvfile.write("Datum,Kurz,Den v týdnu,Svátek,Poznámka\n")
|
||||
|
||||
@@ -149,7 +386,9 @@ def generate_yearly_report(year, currency_code, output_dir="data"):
|
||||
pass
|
||||
else:
|
||||
# Kurz nebyl nalezen, zkusíme dopočítat
|
||||
calculated_rate = get_rate_for_date_with_fallback(date_str, currency_code)
|
||||
calculated_rate = get_rate_for_date_with_fallback(
|
||||
date_str, currency_code
|
||||
)
|
||||
if calculated_rate is not None:
|
||||
rate = calculated_rate
|
||||
note = "Dopočítaný kurz"
|
||||
@@ -160,7 +399,9 @@ def generate_yearly_report(year, currency_code, output_dir="data"):
|
||||
note = "Kurz není k dispozici"
|
||||
|
||||
# Zapišeme řádek do CSV
|
||||
csvfile.write(f"{date_str},{rate if rate is not None else ''},{day_name},{holiday_text},{note}\n")
|
||||
csvfile.write(
|
||||
f"{date_str},{rate if rate is not None else ''},{day_name},{holiday_text},{note}\n"
|
||||
)
|
||||
|
||||
# Přejdeme na další den
|
||||
current_date += timedelta(days=1)
|
||||
@@ -171,6 +412,7 @@ def generate_yearly_report(year, currency_code, output_dir="data"):
|
||||
debug_print(f"Chyba při zápisu do souboru: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def generate_monthly_report(year, month, currency_code, output_dir="data"):
|
||||
"""
|
||||
Vygeneruje report kurzů pro zadaný měsíc, rok a měnu.
|
||||
@@ -212,7 +454,7 @@ def generate_monthly_report(year, month, currency_code, output_dir="data"):
|
||||
filepath = os.path.join(output_dir, filename)
|
||||
|
||||
try:
|
||||
with open(filepath, 'w', newline='', encoding='utf-8') as csvfile:
|
||||
with open(filepath, "w", newline="", encoding="utf-8") as csvfile:
|
||||
# Nové pořadí sloupců: Datum, Kurz, Den v týdnu, Svátek, Poznámka
|
||||
csvfile.write("Datum,Kurz,Den v týdnu,Svátek,Poznámka\n")
|
||||
|
||||
@@ -236,7 +478,9 @@ def generate_monthly_report(year, month, currency_code, output_dir="data"):
|
||||
pass
|
||||
else:
|
||||
# Kurz nebyl nalezen, zkusíme dopočítat
|
||||
calculated_rate = get_rate_for_date_with_fallback(date_str, currency_code)
|
||||
calculated_rate = get_rate_for_date_with_fallback(
|
||||
date_str, currency_code
|
||||
)
|
||||
if calculated_rate is not None:
|
||||
rate = calculated_rate
|
||||
note = "Dopočítaný kurz"
|
||||
@@ -247,7 +491,9 @@ def generate_monthly_report(year, month, currency_code, output_dir="data"):
|
||||
note = "Kurz není k dispozici"
|
||||
|
||||
# Zapišeme řádek do CSV
|
||||
csvfile.write(f"{date_str},{rate if rate is not None else ''},{day_name},{holiday_text},{note}\n")
|
||||
csvfile.write(
|
||||
f"{date_str},{rate if rate is not None else ''},{day_name},{holiday_text},{note}\n"
|
||||
)
|
||||
|
||||
# Přejdeme na další den
|
||||
current_date += timedelta(days=1)
|
||||
@@ -258,7 +504,10 @@ def generate_monthly_report(year, month, currency_code, output_dir="data"):
|
||||
debug_print(f"Chyba při zápisu do souboru: {e}")
|
||||
return None
|
||||
|
||||
def generate_period_report(start_date_str, end_date_str, currency_code, output_dir="data"):
|
||||
|
||||
def generate_period_report(
|
||||
start_date_str, end_date_str, currency_code, output_dir="data"
|
||||
):
|
||||
"""
|
||||
Vygeneruje report kurzů pro zadané období a měnu.
|
||||
|
||||
@@ -268,7 +517,9 @@ def generate_period_report(start_date_str, end_date_str, currency_code, output_d
|
||||
:param output_dir: Adresář, kam se má CSV soubor s reportem uložit.
|
||||
:return: Chta k vytvořenému CSV souboru.
|
||||
"""
|
||||
debug_print(f"Generuji report kurzů pro {currency_code} za období {start_date_str} - {end_date_str}...")
|
||||
debug_print(
|
||||
f"Generuji report kurzů pro {currency_code} za období {start_date_str} - {end_date_str}..."
|
||||
)
|
||||
|
||||
try:
|
||||
start_date = datetime.strptime(start_date_str, "%d.%m.%Y")
|
||||
@@ -281,12 +532,16 @@ def generate_period_report(start_date_str, end_date_str, currency_code, output_d
|
||||
today = datetime.now()
|
||||
# Pokud je požadované období v budoucnosti, nepokračujeme
|
||||
if start_date.date() > today.date():
|
||||
debug_print(f"Chyba: Nelze generovat report pro období v budoucnosti (od {start_date_str}).")
|
||||
debug_print(
|
||||
f"Chyba: Nelze generovat report pro období v budoucnosti (od {start_date_str})."
|
||||
)
|
||||
return None
|
||||
|
||||
# Pokud je konec období v budoucnosti, omezíme ho na dnešní datum
|
||||
if end_date.date() > today.date():
|
||||
debug_print(f"Upozornění: Konec období byl omezen na dnešní datum ({today.strftime('%d.%m.%Y')}), protože zbytek je v budoucnosti.")
|
||||
debug_print(
|
||||
f"Upozornění: Konec období byl omezen na dnešní datum ({today.strftime('%d.%m.%Y')}), protože zbytek je v budoucnosti."
|
||||
)
|
||||
end_date = today
|
||||
|
||||
# Zkontrolujeme, zda databáze obsahuje data pro roky v rozmezí
|
||||
@@ -298,14 +553,14 @@ def generate_period_report(start_date_str, end_date_str, currency_code, output_d
|
||||
debug_print(f"Stahuji roční data pro rok {year}...")
|
||||
# Stáhneme roční data s vynuceným stažením
|
||||
os.makedirs("data", exist_ok=True)
|
||||
data_fetcher.download_yearly_data(year, output_dir="data", force=True)
|
||||
data_fetcher.download_yearly_data(year, output_dir="data")
|
||||
|
||||
# Otevřeme CSV soubor pro zápis
|
||||
filename = f"report_{currency_code}_{start_date_str.replace('.', '_')}_to_{end_date_str.replace('.', '_')}.csv"
|
||||
filepath = os.path.join(output_dir, filename)
|
||||
|
||||
try:
|
||||
with open(filepath, 'w', newline='', encoding='utf-8') as csvfile:
|
||||
with open(filepath, "w", newline="", encoding="utf-8") as csvfile:
|
||||
# Nové pořadí sloupců: Datum, Kurz, Den v týdnu, Svátek, Poznámka
|
||||
csvfile.write("Datum,Kurz,Den v týdnu,Svátek,Poznámka\n")
|
||||
|
||||
@@ -329,7 +584,9 @@ def generate_period_report(start_date_str, end_date_str, currency_code, output_d
|
||||
pass
|
||||
else:
|
||||
# Kurz nebyl nalezen, zkusíme dopočítat
|
||||
calculated_rate = get_rate_for_date_with_fallback(date_str, currency_code)
|
||||
calculated_rate = get_rate_for_date_with_fallback(
|
||||
date_str, currency_code
|
||||
)
|
||||
if calculated_rate is not None:
|
||||
rate = calculated_rate
|
||||
note = "Dopočítaný kurz"
|
||||
@@ -340,7 +597,9 @@ def generate_period_report(start_date_str, end_date_str, currency_code, output_d
|
||||
note = "Kurz není k dispozici"
|
||||
|
||||
# Zapišeme řádek do CSV
|
||||
csvfile.write(f"{date_str},{rate if rate is not None else ''},{day_name},{holiday_text},{note}\n")
|
||||
csvfile.write(
|
||||
f"{date_str},{rate if rate is not None else ''},{day_name},{holiday_text},{note}\n"
|
||||
)
|
||||
|
||||
# Přejdeme na další den
|
||||
current_date += timedelta(days=1)
|
||||
@@ -351,29 +610,6 @@ def generate_period_report(start_date_str, end_date_str, currency_code, output_d
|
||||
debug_print(f"Chyba při zápisu do souboru: {e}")
|
||||
return None
|
||||
|
||||
def get_czech_day_name(date_str):
|
||||
"""
|
||||
Vrátí český název dne v týdnu pro zadané datum.
|
||||
|
||||
:param date_str: Datum ve formátu DD.MM.YYYY
|
||||
:return: Český název dne v týdnu
|
||||
"""
|
||||
try:
|
||||
date_obj = datetime.strptime(date_str, "%d.%m.%Y")
|
||||
# Czech day names
|
||||
czech_days = ["pondělí", "úterý", "středa", "čtvrtek", "pátek", "sobota", "neděle"]
|
||||
return czech_days[date_obj.weekday()]
|
||||
except ValueError:
|
||||
return "neznámý den"
|
||||
|
||||
def is_holiday(date_str):
|
||||
"""
|
||||
Zkontroluje, zda je zadané datum státní svátek.
|
||||
|
||||
:param date_str: Datum ve formátu DD.MM.YYYY
|
||||
:return: True pokud je svátek, jinak False
|
||||
"""
|
||||
return holidays.is_holiday(date_str)
|
||||
|
||||
# Příklad použití
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user