Python实现Oracle数据库同步

admin

1. # -*- coding: utf-8 -*-
   
2. from WindPy import *
   
3. from datetime import datetime, timedelta
   
4. import time
   
5. import cx_Oracle
   
6. import sys,traceback
   
7. import os
   
8. os.environ['NLS_LANG'] = 'SIMPLIFIED CHINESE_CHINA.UTF8'
   
9. 
   
10. year = int(datetime.now().strftime('%Y'))
    
11. today = int(datetime.now().strftime('%Y%m%d'))
    
12. dayOfWeek = datetime.strptime(str(today),"%Y%m%d").date().weekday() + 1
    
13. print("today:%d,dayOfWeek:%d" %(today,dayOfWeek))
    
14. 
    
15. def getDataFromSjzx(today,sql,param,callFunName):
    
16.   conn_sjzx = cx_Oracle.connect('stage/stage@10.201.200.29:1521/orcl')
    
17.   cur_sjzx = conn_sjzx.cursor()
    
18.   cur_sjzx.prepare(sql)
    
19.   print("[getDataFromSjzx---%s]传入参数:%s" % (callFunName,param))
    
20.   print("[getDataFromSjzx---%s]传入Sql语句:%s" % (callFunName,sql))
    
21.   print("[getDataFromSjzx---%s]ips绑定变量:%s" % (callFunName,cur_sjzx.bindnames()))
    
22.   cur_sjzx.execute(None, param)
    
23.   fromData = cur_sjzx.fetchall()
    
24.   desc = cur_sjzx.description
    
25.   cur_sjzx.close()
    
26.   conn_sjzx.close()
    
27.   length = len(desc)
    
28.   dataListFromSjzx = []
    
29.   for data in fromData:
    
30.     row = {}
    
31.     for i in range(length):
    
32.       if data[i] is not None:
    
33.         row[desc[i][0].lower()] = data[i]
    
34.     dataListFromSjzx.append(row)
    
35.   print("[getDataFromSjzx---%s]日期查询[%d]总记录数：%d" % (callFunName ,today, len(dataListFromSjzx)))
    
36.   return dataListFromSjzx
    
37. 
    
38. def updateDataToIps(sql,param,callFunName):
    
39.   conn_ips = cx_Oracle.connect('ips/ips@10.201.200.42:1521/xjsdb')
    
40.   cursor_ips = conn_ips.cursor()
    
41.   cursor_ips.prepare(sql)
    
42.   #print("[updateDataToIps]传入参数:%s" % (param))
    
43.   print("[updateDataToIps---%s]传入Sql语句:%s" % (callFunName,sql))
    
44.   print("[updateDataToIps---%s]ips绑定变量:%s" % (callFunName,cursor_ips.bindnames()))
    
45.   cursor_ips.executemany(None, param)
    
46.   conn_ips.commit()
    
47.   cursor_ips.close()
    
48.   conn_ips.close()
    
49. 
    
50. def updateDeleteDataToIps(delete_sql,delete_param,insert_sql,insert_param,callFunName):
    
51.   conn_ips = cx_Oracle.connect('ips/ips@10.201.200.42:1521/xjsdb')
    
52.   cursor_ips = conn_ips.cursor()
    
53.   cursor_ips.prepare(delete_sql)
    
54.   print("[updateDataToIps---%s]传入Sql语句(删除):%s" % (callFunName,delete_sql))
    
55.   print("[updateDataToIps---%s]ips绑定变量(删除):%s" % (callFunName,cursor_ips.bindnames()))
    
56.   cursor_ips.execute(None,delete_param)
    
57.   cursor_ips.prepare(insert_sql)
    
58.   print("[updateDataToIps---%s]传入Sql语句(插入):%s" % (callFunName,insert_sql))
    
59.   print("[updateDataToIps---%s]ips绑定变量(插入):%s" % (callFunName,cursor_ips.bindnames()))
    
60.   cursor_ips.executemany(None, insert_param)
    
61.   conn_ips.commit()
    
62.   cursor_ips.close()
    
63.   conn_ips.close()
    
64. 
    
65. def synTradeDayData():
    
66.   select_sql = "select t.trade_days, t.s_info_exchmarket " \
    
67.                "  from wind2_asharecalendar t " \
    
68.                " where to_number(t.trade_days) >= :today " \
    
69.                "   and t.s_info_exchmarket = :marketType " \
    
70.                " order by to_number(t.trade_days) "
    
71.   param = {}
    
72.   marketType = "SSE"
    
73.   param["today"] = today
    
74.   param["marketType"] = marketType
    
75.   dataListFromSjzx = []
    
76.   dataListFromSjzx = getDataFromSjzx(today, select_sql, param, "synTradeDayData")
    
77.   dataListToIps = []
    
78.   dataListToIps = fillTradeDay(dataListFromSjzx)
    
79.   insert_sql = "insert into t_trade_day (l_trade_date,vc_market_type) values (:tradeDay,:marketType)"
    
80.   delete_sql = "delete from t_trade_day a where a.l_trade_date >= :today and a.vc_market_type = :marketType"
    
81.   if (dataListToIps is not None and len(dataListToIps) > 0):
    
82.     updateDeleteDataToIps(delete_sql,param,insert_sql,dataListToIps,"synTradeDayData")
    
83. 
    
84. def fillTradeDay(dataListFromSjzx):
    
85.   dataListToIps = []
    
86.   for data in dataListFromSjzx:
    
87.     result = {}
    
88.     result["tradeDay"] = data['trade_days']
    
89.     result["marketType"] = data['s_info_exchmarket']
    
90.     dataListToIps.append(result)
    
91.   return dataListToIps
    
92. 
    
93. def synSpecialNonTradedayData():
    
94.   if (dayOfWeek==6 or dayOfWeek==7):
    
95.     print("[synSpecialNonTradedayData]日期[%d]为周%d,无需进行同步处理" % (today,dayOfWeek))
    
96.     return
    
97.   select_sql = "select nvl(count(1),0) l_count " \
    
98.                "  from wind2_asharecalendar t " \
    
99.                " where to_number(t.trade_days) = :today " \
    
100.                "   and t.s_info_exchmarket = :marketType "
     
101.   param = {}
     
102.   marketType = "SSE"
     
103.   param["today"] = today
     
104.   param["marketType"] = marketType
     
105.   dataListFromSjzx = []
     
106.   dataListFromSjzx = getDataFromSjzx(today, select_sql, param, "synSpecialNonTradedayData")
     
107.   if (dataListFromSjzx is not None and dataListFromSjzx[0]["l_count"] > 0):
     
108.     print("[synSpecialNonTradedayData]日期[%d]为周%d且为交易日,无需进行同步处理" % (today, dayOfWeek))
     
109.     return
     
110.   dataListToIps = []
     
111.   dataListToIps = fillSpecialNonTradeday(dataListFromSjzx)
     
112.   delete_sql = "delete from t_special_non_tradeday a where a.l_trade_date >= :today and a.vc_market_type = :marketType"
     
113.   insert_sql = "insert into t_special_non_tradeday (l_trade_date,vc_market_type,l_day_week) values (:tradeDay,:marketType,:dayOfWeek)"
     
114.   if (dataListToIps is not None and len(dataListToIps) > 0):
     
115.     updateDeleteDataToIps(delete_sql, param, insert_sql, dataListToIps, "synSpecialNonTradedayData")
     
116. 
     
117. def fillSpecialNonTradeday(dataListFromSjzx):
     
118.   dataListToIps = []
     
119.   result = {}
     
120.   result["tradeDay"] = today
     
121.   result["marketType"] = "SSE"
     
122.   result["dayOfWeek"] = dayOfWeek
     
123.   dataListToIps.append(result)
     
124.   return dataListToIps
     
125. 
     
126. def synStockInfoWind():
     
127.   select_sql = "select a.ob_object_id      vc_object_id, " \
     
128.                "       a.f1_0001           vc_wind_code, " \
     
129.                "       a.f4_0001           vc_stock_code, " \
     
130.                "       a.f6_0001           vc_stock_name, " \
     
131.                "       a.f10_0001          vc_type_wind, " \
     
132.                "       a.f11_0001          vc_type_detail_wind, " \
     
133.                "       a.f12_0001          vc_stock_type, " \
     
134.                "       a.f14_0001          vc_market_type, " \
     
135.                "       a.f16_0001          vc_stock_id_wind, " \
     
136.                "       a.f17_0001          vc_company_id_wind, " \
     
137.                "       to_number(b.s_info_listdate)   l_listing_date, " \
     
138.                "       nvl(to_number(b.s_info_delistdate),99999999) l_delisting_date " \
     
139.                "  from wind_tb_object_0001 a, wind2_asharedescription b " \
     
140.                " where a.f12_0001 = :stockType " \
     
141.                "   and a.f1_0001 = b.s_info_windcode " \
     
142.                "   and b.s_info_listdate is not null " \
     
143.                "   and (to_number(b.s_info_listdate) >= :today or " \
     
144.                "       (b.s_info_delistdate is not null and to_number(b.s_info_delistdate) >= :today)) "
     
145.   param = {}
     
146.   stockType = "A"
     
147.   param["today"] = today
     
148.   param["stockType"] = stockType
     
149.   dataListFromSjzx = []
     
150.   dataListFromSjzx = getDataFromSjzx(today, select_sql, param, "synStockInfoWind")
     
151.   dataListToIps = []
     
152.   deleteStockCodeStr = ""
     
153.   (dataListToIps,deleteStockCodeStr) = fillStockInfoWind(dataListFromSjzx)
     
154.   print("[synStockInfoWind]待删除数据:%s" % (deleteStockCodeStr))
     
155.   delete_sql = "delete from t_stock_info_wind where vc_wind_code in ("+deleteStockCodeStr+")"
     
156.   insert_sql = "insert into t_stock_info_wind ( "\
     
157.                "    vc_object_id ,vc_wind_code ,vc_stock_code ,vc_stock_name , "\
     
158.                "    vc_type_wind ,vc_type_detail_wind ,vc_stock_type ,vc_market_type , "\
     
159.                "    vc_stock_id_wind ,vc_company_id_wind ,l_listing_date ,l_delisting_date) "\
     
160.                "  values (:objectId,:windCode,:stockCode,:stockName, "\
     
161.                "    :typeWind,:typeDetailWind,:stockType,:marketType, "\
     
162.                "    :stockIdWind,:companyIdWind,:listingDate,:delistingDate) "
     
163.   if (dataListToIps is not None and len(dataListToIps) > 0):
     
164.     if (deleteStockCodeStr is not None and len(deleteStockCodeStr) > 0):
     
165.       updateDeleteDataToIps(delete_sql, {}, insert_sql, dataListToIps, "synStockInfoWind")
     
166.     else:
     
167.       updateDataToIps(insert_sql, dataListToIps, "synStockInfoWind")
     
168. 
     
169. def fillStockInfoWind(dataListFromSjzx):
     
170.   dataListToIps = []
     
171.   deleteStockCodeStr = ""
     
172.   for data in dataListFromSjzx:
     
173.     result = {}
     
174.     result["objectId"] = data["vc_object_id"]
     
175.     result["windCode"] = data["vc_wind_code"]
     
176.     result["stockCode"] = data["vc_stock_code"]
     
177.     result["stockName"] = data["vc_stock_name"]
     
178.     result["typeWind"] = data["vc_type_wind"]
     
179.     result["typeDetailWind"] = data["vc_type_detail_wind"]
     
180.     result["stockType"] = data["vc_stock_type"]
     
181.     result["marketType"] = data["vc_market_type"]
     
182.     result["stockIdWind"] = data["vc_stock_id_wind"]
     
183.     result["companyIdWind"] = data["vc_company_id_wind"]
     
184.     result["listingDate"] = data["l_listing_date"]
     
185.     result["delistingDate"] = data["l_delisting_date"]
     
186.     if (result["delistingDate"] is not None and result["delistingDate"] > 0 and result["delistingDate"] < 99999999):
     
187.       if (deleteStockCodeStr is None or len(deleteStockCodeStr) <= 0):
     
188.         deleteStockCodeStr = "'" + result["windCode"] + "'"
     
189.       else:
     
190.         deleteStockCodeStr = deleteStockCodeStr + "," + "'" + result["windCode"] + "'"
     
191.     dataListToIps.append(result)
     
192.   return (dataListToIps,deleteStockCodeStr)
     
193. 
     
194. def synWind2AshareeodpricesData():
     
195.   table = "wind2_ashareeodprices_" + str(year)
     
196.   select_sql = "select a.object_id,a.s_info_windcode,nvl(to_number(a.trade_dt),0) trade_dt,a.crncy_code, " \
     
197.                "       nvl(to_number(a.s_dq_preclose),0) s_dq_preclose, " \
     
198.                "       nvl(to_number(a.s_dq_open),0) s_dq_open, " \
     
199.                "       nvl(to_number(a.s_dq_close),0) s_dq_close, " \
     
200.                "       nvl(to_number(a.s_dq_volume),0) s_dq_volume, " \
     
201.                "       nvl(to_number(a.s_dq_amount),0) s_dq_amount, " \
     
202.                 "      a.s_dq_tradestatus, " \
     
203.                "       a.l_createdate,a.l_createtime " \
     
204.                "  from " + table + " a " \
     
205.                " where to_number(a.trade_dt) >= :today"
     
206.   param = {}
     
207.   param["today"] = today
     
208.   dataListFromSjzx = []
     
209.   dataListFromSjzx = getDataFromSjzx(today,select_sql,param,"synWind2AshareeodpricesData")
     
210.   dataListToIps = []
     
211.   dataListToIps = fillWind2AshareeodpricesData(dataListFromSjzx)
     
212.   insert_sql = "insert into " + table + "_bak ( " \
     
213.                "         object_id,s_info_windcode,trade_dt,crncy_code, " \
     
214.                "         s_dq_preclose,s_dq_open,s_dq_close,s_dq_volume,s_dq_amount, " \
     
215.                "         s_dq_tradestatus,l_createdate,l_createtime) " \
     
216.                "  values(:objectId,:windcode,:tradeDt,:crncyCode, " \
     
217.                "         :dqPreclose,:dqOpen,:dqClose,:dqVolume,:dqAmount, " \
     
218.                "         :dqTradestatus,:createDate,:createTime)"
     
219.   if (dataListToIps is not None and len(dataListToIps) > 0):
     
220.     updateDataToIps(insert_sql, dataListToIps, "synWind2AshareeodpricesData")
     
221. 
     
222. def fillWind2AshareeodpricesData(dataListFromSjzx):
     
223.   dataListToIps = []
     
224.   for data in dataListFromSjzx:
     
225.     result = {}
     
226.     result["objectId"] = data["object_id"]
     
227.     result["windcode"] = data["s_info_windcode"]
     
228.     result["tradeDt"] = data["trade_dt"]
     
229.     result["crncyCode"] = data["crncy_code"]
     
230.     result["dqPreclose"] = data["s_dq_preclose"]
     
231.     result["dqOpen"] = data["s_dq_open"]
     
232.     result["dqClose"] = data["s_dq_close"]
     
233.     result["dqVolume"] = data["s_dq_volume"]
     
234.     result["dqAmount"] = data["s_dq_amount"]
     
235.     result["dqTradestatus"] = data["s_dq_tradestatus"]
     
236.     result["createDate"] = data["l_createdate"]
     
237.     result["createTime"] = data["l_createtime"]
     
238.     dataListToIps.append(result)
     
239.   return dataListToIps
     
240. 
     
241. def synAshareDividendData():
     
242.   select_sql = "select a.s_info_windcode vc_wind_code, "\
     
243.                "       a.ex_dt l_ex_date, "\
     
244.                "       sum(a.cash_dvd_per_sh_pre_tax) en_cash_dvd_per_sh_pre_tax, "\
     
245.                "       sum(a.cash_dvd_per_sh_after_tax) en_cash_dvd_per_sh_after_tax "\
     
246.                "   from wind2_asharedividend a "\
     
247.                " where a.ex_dt is not null "\
     
248.                "   and to_number(a.ex_dt) >= :today "\
     
249.                "   group by a.s_info_windcode, a.ex_dt order by a.ex_dt desc"
     
250.   param = {}
     
251.   param["today"] = today
     
252.   dataListFromSjzx = []
     
253.   dataListFromSjzx = getDataFromSjzx(today, select_sql, param, "synAshareDividendData")
     
254.   dataListToIps = []
     
255.   dataListToIps = fillAshareDividendData(dataListFromSjzx)
     
256.   insert_sql = "insert into t_ashare_dividend ( "\
     
257.                "         vc_wind_code,l_ex_date,en_cash_dvd_per_sh_pre_tax,en_cash_dvd_per_sh_after_tax) "\
     
258.                "  values(:windCode,:exDate,:cashDvdPerShPreTax,:cashDvdPerShAfterTax)"
     
259.   if (dataListToIps is not None and len(dataListToIps) > 0):
     
260.     updateDataToIps(insert_sql, dataListToIps, "synAshareDividendData")
     
261. 
     
262. def fillAshareDividendData(dataListFromSjzx):
     
263.   dataListToIps = []
     
264.   for data in dataListFromSjzx:
     
265.     result = {}
     
266.     result["windCode"] = data["vc_wind_code"]
     
267.     result["exDate"] = data["l_ex_date"]
     
268.     result["cashDvdPerShPreTax"] = data["en_cash_dvd_per_sh_pre_tax"]
     
269.     result["cashDvdPerShAfterTax"] = data["en_cash_dvd_per_sh_after_tax"]
     
270.     dataListToIps.append(result)
     
271.   return dataListToIps
     
272. 
     
273. def synCapitalStockWindData():
     
274.   select_sql = "select capital.vc_wind_code, " \
     
275.                "       capital.l_change_date, " \
     
276.                "       capital.l_capital_type, " \
     
277.                "       capital.en_captial_value " \
     
278.                "  from (select b.f1_0001 vc_wind_code, " \
     
279.                "               to_number(a.f50_1432) l_change_date, "\
     
280.                "               '1' l_capital_type, " \
     
281.                "               a.f27_1432 en_captial_value " \
     
282.                "          from wind_tb_object_1432 a, wind_tb_object_0001 b " \
     
283.                "        where a.f1_1432 = b.f17_0001 " \
     
284.                "          and b.f12_0001 = 'A' " \
     
285.                "          and a.ob_is_valid_1432 = '1' " \
     
286.                "          and to_number(a.f50_1432) >= :today " \
     
287.                "        union all " \
     
288.                "        select d.f1_0001 vc_wind_code, " \
     
289.                "               to_number(c.f2_1931) l_change_date, "\
     
290.                "               '2' l_capital_type, " \
     
291.                "               c.f4_1931 en_captial_value " \
     
292.                "          from wind_tb_object_1931 c, wind_tb_object_0001 d " \
     
293.                "        where c.f1_1931 = d.f16_0001 " \
     
294.                "          and d.f12_0001 = 'A' " \
     
295.                "          and to_number(c.f2_1931) >= :today) capital " \
     
296.                "  order by capital.l_change_date"
     
297.   param = {}
     
298.   param["today"] = today
     
299.   dataListFromSjzx = []
     
300.   dataListFromSjzx = getDataFromSjzx(today, select_sql, param, "synCapitalStockWindData")
     
301.   dataListToIps = []
     
302.   dataListToIps = fillCapitalStockWindData(dataListFromSjzx)
     
303.   insert_sql = "insert into t_capital_stock_wind ( "\
     
304.                "         vc_wind_code,l_change_date,l_capital_type,en_captial_value) "\
     
305.                "  values(:windCode,:changeDate,:capitalType,:captialValue)"
     
306.   if (dataListToIps is not None and len(dataListToIps) > 0):
     
307.     updateDataToIps(insert_sql, dataListToIps, "synCapitalStockWindData")
     
308. 
     
309. def fillCapitalStockWindData(dataListFromSjzx):
     
310.   dataListToIps = []
     
311.   for data in dataListFromSjzx:
     
312.     result = {}
     
313.     result["windCode"] = data["vc_wind_code"]
     
314.     result["changeDate"] = data["l_change_date"]
     
315.     result["capitalType"] = data["l_capital_type"]
     
316.     result["captialValue"] = data["en_captial_value"]
     
317.     dataListToIps.append(result)
     
318.   return dataListToIps
     
319. 
     
320. def synSwIndustryClass():
     
321.   select_sql = "select c.f1_0001 vc_wind_code, " \
     
322.                "       c.f6_0001 vc_stock_name, " \
     
323.                "       to_number(a.f3_1476) l_import_date, " \
     
324.                "       to_number(nvl(a.f4_1476,0)) l_remove_date, " \
     
325.                "       a.f5_1476 l_new_flag, " \
     
326.                "       b.levelnum-1 l_industry_level, " \
     
327.                "       (case when b.levelnum=2 then substr(a.f2_1476,1,4) " \
     
328.                "             when b.levelnum=3 then substr(a.f2_1476,1,6) " \
     
329.                "             when b.levelnum=4 then substr(a.f2_1476,1,8) " \
     
330.                "        else '' end) vc_industry_code, " \
     
331.                "       b.name vc_industry_name " \
     
332.                "  from wind_tb_object_1476 a, wind_tb_object_1022 b, wind_tb_object_0001 c " \
     
333.                " where b.used = '1' " \
     
334.                "   and b.code like '61%' " \
     
335.                "   and c.f12_0001 = 'A' " \
     
336.                "   and a.f1_1476 = c.f16_0001 " \
     
337.                "   and (to_number(a.f3_1476) >= :today or  " \
     
338.                "        to_number(nvl(a.f4_1476,0)) >= :today) " \
     
339.                "   and ((b.levelnum = 2 and substr(b.code,1,4) = substr(a.f2_1476,1,4)) or  " \
     
340.                "        (b.levelnum = 3 and substr(b.code,1,6) = substr(a.f2_1476,1,6)) or  " \
     
341.                "        (b.levelnum = 4 and substr(b.code,1,8) = substr(a.f2_1476,1,8))) " \
     
342.                "   order by b.levelnum "
     
343.   param = {}
     
344.   param["today"] = today
     
345.   dataListFromSjzx = []
     
346.   dataListFromSjzx = getDataFromSjzx(today, select_sql, param, "synSwIndustryClass")
     
347.   dataListToIps = []
     
348.   dataListToIps = fillSwIndustryClass(dataListFromSjzx)
     
349.   insert_sql = "insert into t_sw_industry_class ( "\
     
350.                "         vc_wind_code,vc_stock_name,l_import_date,l_remove_date, "\
     
351.                "         l_new_flag,l_industry_level,vc_industry_code,vc_industry_name) "\
     
352.                "  values(:windCode,:stockName,:importDate,:removeDate, "\
     
353.                "         :newFlag,:industryLevel,:industryCode,:industryName)"
     
354.   if (dataListToIps is not None and len(dataListToIps) > 0):
     
355.     updateDataToIps(insert_sql, dataListToIps, "synSwIndustryClass")
     
356. 
     
357. def fillSwIndustryClass(dataListFromSjzx):
     
358.   dataListToIps = []
     
359.   for data in dataListFromSjzx:
     
360.     result = {}
     
361.     result["windCode"] = data["vc_wind_code"]
     
362.     result["stockName"] = data["vc_stock_name"]
     
363.     result["importDate"] = data["l_import_date"]
     
364.     result["removeDate"] = data["l_remove_date"]
     
365.     result["newFlag"] = data["l_new_flag"]
     
366.     result["industryLevel"] = data["l_industry_level"]
     
367.     result["industryCode"] = data["vc_industry_code"]
     
368.     result["industryName"] = data["vc_industry_name"]
     
369.     dataListToIps.append(result)
     
370.   return dataListToIps
     
371. 
     
372. if __name__ == "__main__":
     
373.   synTradeDayData()
     
374.   synSpecialNonTradedayData()
     
375.   synStockInfoWind()
     
376.   synWind2AshareeodpricesData()
     
377.   synAshareDividendData()
     
378.   synCapitalStockWindData()
     
379.   synSwIndustryClass()
     

复制代码